rabbitmq简介与基本使用
rabbitmq简介与基本使用
RabbitMQ 是轻量级的，易于在本地和云端部署。它支持多种消息传递协议。RabbitMQ 可以部署在分布式和联合配置中，以满足大规模、高可用性的要求。
一、RabbitMQ的结构
rabbitmq遵循AMQP协议。
-
Broker:接收和分发消息的应用，RabbitMQ就是MessageBroker
-
Virtual Host:虚拟Broker，将多个单元隔离开
-
Connection： publisher/consumer和broker之间的tcp连接
-
Channel：connection内部建立的逻辑连接，通常每个线程创建单独的channel
-
Routing Key： 路由键，用来指示消息的路由转发，相当于快递的地址
-
Exchange：交换机，相当于快递的分拨中心
-
Queue： 队列，消息最终被送到这里等待consumer取走
-
Binding：exchange和queue之间的虚拟连接，用于message的分发依据
Exhange是AMQP协议和rabbitmq的核心组件
-
Exchange的功能是根据绑定关系和路由键为消息提供路由，将消息转发值相应的队列
-
exchange有4种类型：Direct/Topic/Fanout/Heders,其中Headers使用很少，以前三种为主
-
Direct(直接路由)：Routing Key =Binding Key，容易配置使用
-
Fanout（广播路由）：群发绑定的所有队列，使用与消息广播
-
Topic（话题路由）：功能较为复杂，但使用灵活，建议优先使用，为以后拓展留余地。
二、应用场景
1、流量削峰
举个例子，如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。
2.应用解耦
以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，中单用户感受不到物流系统的故障，提升系统的可用性。
3.异步处理
有些服务间调用是异步的，例如 A 调用 B，B 需要花费很长时间执行，但是 A 需要知道 B 什么时候可以执行完，以前一般有两种方式，A 过一段时间去调用 B 的查询 api 查询。或者 A 提供一个callback api，B 执行完之后调用 api 通知 A 服务。这两种方式都不是很优雅，使用消息总线，可以很方便解决这个问题，A 调用 B 服务后，只需要监听 B 处理完成的消息，当 B 处理完成后，会发送一条消息给 MQ，MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api，也不用提供 callback api。同样 B 服务也不用做这些操作。A 服务还能及时的得到异步处理成功的消息。
三、基本使用
3.1 生产者
static String QUEUE = "restaurant";
static String EXCHANGE = "exchange.order.restaurant";
@Test
public void mqTestConnect() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
factory.setHost("192.168.6.131");
factory.setPort(5672);
try (
Connection conn = factory.newConnection();
Channel channel = conn.openChannel().get()
){
/**
* 生成一个交换机
* 1.交换机名称
* 2.交换机里面的消息是否持久化 默认消息存储在内存中
* 3.是否自动删除 如果服务器在不再使用交换机时删除该交换机
* 4.其他参数
*/
channel.exchangeDeclare(
EXCHANGE,
BuiltinExchangeType.DIRECT,
true,
false,
null
);
/**
* 生成一个队列
* 1.队列名称
* 2.队列里面的消息是否持久化 默认消息存储在内存中
* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
* 5.其他参数
*/
channel.queueDeclare(QUEUE, false, false, false, null);
/**
* 绑定交换机与队列
*/
channel.queueBind(QUEUE,EXCHANGE,"key.order");
String payload = "msg";
/**
* 发送一个消息
* 1.发送到那个交换机
* 2.路由的 key 是哪个
* 3.其他的参数信息
* 4.发送消息的消息体
*/
channel.basicPublish(EXCHANGE,"key.order",null,payload.getBytes());
}
}
rabbitmq控制台就会有对应的交换机与队列以及消息
绑定信息
消息
3.2 消费者代码
@Test
public void mqConsumer() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
factory.setHost("192.168.6.131");
factory.setPort(5672);
try (
Connection conn = factory.newConnection();
Channel channel = conn.openChannel().get()
){
log.info("start listening message...");
/**
* 消费者消费消息
* 1.消费哪个队列
* 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
* 3.消费者未成功消费的回调
*/
channel.basicConsume(QUEUE, deliverCallback, consumerTag -> {
log.info("消息消费被中断");
});
while (true) {
Thread.sleep(100000);
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
DeliverCallback deliverCallback = ((consumerTag, message) -> {
String messageBody = new String(message.getBody());
log.info("deliverCallback:messageBody:{}", messageBody);
});
运行后，控制台打印
