Java真的不难（五十四）RabbitMQ的入门及使用

RabbitMQ的入门及使用

一、什么是RabbitMQ？

MQ全称为Message Queue，即消息队列。消息队列是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦

二、RabbitMQ与Kafka的全面对比

对比项KafkaRabbitMQ开发语言scala、Javaerlang是否支持多租户2.x.x支持支持是否支持topic优先级不支持支持是否支持消息全局有序不支持支持是否支持消息分区有序支持支持是否有内置监控无有是否支持多个生产者支持支持是否支持多个消费者支持支持是否支持一个分区多个消费者不支持不支持是否支持JMX支持不支持是否支持加密支持支持消息队列协议支持仅支持自定义协议支持AMQP、MQTT、STOMP协议客户端语言支持支持支持是否支持消息追踪不支持支持是否支持消费者推模式不支持支持否支持消费者拉模式支持支持是否支持广播消息支持支持是否支持消息回溯支持消息回溯，因为消息持久化，消息被消费后会记录offset和timstamp不支持，消息确认被消费后，会被删除是否支持消息数据持久化支持支持是否支持流量控制支持支持是否支持事务性消息支持不支持元数据管理通过zookeeper进行管理支持默认服务端口90925672默认监控端口kafka web console 9000;kafka manager 9000;15672相对网络开销较小较大相对内存消耗较小较大相对cpu消耗较大较小
实际场景选择：

Kafka ：
常作为消息传输的数据管道 ，优势主要体现在吞吐量上，虽然可以通过策略实现数据不丢失，严谨性上不如 RabbitMQ，但 kafka保证每条消息最少送达一次，有较小的概率会出现数据重复发送的情况 ，若消息吞吐量极大则Kafka

RabbitMQ：
RabbitMQ金融场景中经常使用 ，常作为交易数据作为数据传输管道， 具有较高的严谨性，数据丢失的可能性更小，具备更高的实时性，和Spring是统一厂商开发，后期支持比较好，目前最流行的，对容错性的处理比较完善

RabbitMQ 支持发布订阅、轮询分发、公平分发、重发、消息拉取
Kafka 不支持重发、事务

三、Linux上安装RabbitMQ

这次安装将RabbitMQ部署在Linux上，可以在电脑本地安装一台Linux，也可以购买云服务器，若购买云服务器，则需要在安全组内把后面要用到的端口给打开！

1、使用Docker安装（最简单的方法）

拉取RabbitMQ：

以下均为Linux指令

#拉取RabbitMQ：docker pull RabbitMQ

#启动rabbitmqdocker run -d--hostname my-rabbit --name rabbit -p15672:15672 -p5672:5672 rabbitmq

#查看docker目前在运行的容器，是否有rabbitmqdockerps

2、图形化安装插件

#进入运行中的容器dockerexec-it 镜像ID /bin/bash

#rabbitmq图形化安装插件
rabbitmq-plugins enable rabbitmq_management

3、WEB页面开启资源监控

#进入容器dockerexec-it rabbitmq /bin/bash

#切到对应目录cd /etc/rabbitmq/conf.d/

#修改 management_agent.disable_metrics_collector = falseecho management_agent.disable_metrics_collector =false> management_agent.disable_metrics_collector.conf

#退出容器exit#重启容器docker restart rabbitmq

然后在浏览器内输入服务器地址+端口（15672）即可进入WEB管理页面，默认账号密码均为：guest

四、Docker配置RabbitMQ集群

先停止在运行的MQ

docker stop 运行容器ID

#启动三个容器docker run -d--hostname rabbitmq01 --name rabbitmqCluster01 -p15672:15672 -p5672:5672 -p1883:1883 -eRABBITMQ_ERLANG_COOKIE='rabbitmqCookie' rabbitmq

docker run -d--hostname rabbitmq02 --name rabbitmqCluster02 -p15673:15672 -p5673:5672 -p1884:1883 -eRABBITMQ_ERLANG_COOKIE='rabbitmqCookie'--link rabbitmqCluster01:rabbitmq01 rabbitmq

docker run -d--hostname rabbitmq03 --name rabbitmqCluster03 -p15674:15672 -p5674:5672 -p1885:1883 -eRABBITMQ_ERLANG_COOKIE='rabbitmqCookie'--link rabbitmqCluster01:rabbitmq01 --link rabbitmqCluster02:rabbitmq02 rabbitmq

#Erlang Cookie 值必须相同，也就是一个集群内 RABBITMQ_ERLANG_COOKIE 参数的值必须相同。因为 RabbitMQ 是用Erlang实现的，Erlang Cookie 相当于不同节点之间通讯的密钥，Erlang节点通过交换 Erlang Cookie 获得认证#进入第二个容器dockerexec-it rabbitmqCluster02 bash
rabbitmqctl stop_app
rabbitmqctl reset

#加入集群
rabbitmqctl join_cluster --ram rabbit@rabbitmq01
rabbitmqctl start_app
exit#进入第三个容器dockerexec-it rabbitmqCluster03 bash
rabbitmqctl stop_app
rabbitmqctl reset

#加入集群
rabbitmqctl join_cluster --ram rabbit@rabbitmq01
rabbitmqctl start_app
exit#主要参数-p15672:15672 management 界面管理访问端口
-p5672:5672 amqp 访问端口
-p1883:1883 mqtt  访问端口

然后依次运行2、3的步骤安装插件和开启资源监控
上诉15672、15673、15674端口均需要在服务器安全组内开启

完成后即可的WEB页面看到MQ的集群

五、SpringBoot整合使用RabbitMQ

依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

yml文件的配置：

spring:
  rabbitmq:
    username: guest
    password: guest
    virtual-host:/
    host: 通道运行的地址
    port:5672

生产者配置类Config：

模拟美团外卖下单，正常下单后若马上被接单则被这条信息直接消费，若5秒内没人接单，该订单消息将进入加急派单队列（死信队列）

所以我们需要两台交换机和两个通道

@Configurationpublicclass mtRabbitConfig {//正常通道的交换机@BeanpublicFanoutExchangemtExchange(){returnnewFanoutExchange("mt_fanout_exchange",true,false);}//死信通道的交换机@BeanpublicFanoutExchangemtDeadExchange(){returnnewFanoutExchange("mt_fanout_dead_exchange",true,false);}//正常通道，给消息设计过期时间，超过该时间未被消费，则进入指定的mt_fanout_dead_exchange@BeanpublicQueuemtQueue(){Map<String,Object> args =newHashMap<>();
        args.put("x-message-ttl",5000);
        args.put("x-dead-letter-exchange","mt_fanout_dead_exchange");returnnewQueue("mt_queue",true,false,false,args);}@BeanpublicQueuemtDaedQueue(){returnnewQueue("mt_dead_queue",true,false,false);}//交换机与通道绑定@BeanpublicBindingmtBinding(){returnBindingBuilder.bind(mtQueue()).to(mtExchange());}@BeanpublicBindingmtDeadBinding(){returnBindingBuilder.bind(mtDaedQueue()).to(mtDeadExchange());}}

生产者（模拟用户下单）：

@AutowiredprivateRabbitTemplate rabbitTemplate;//模拟美团订单下单，若5秒不接单（消费）,则进入死信队列（加急派单）publicvoidmtTakeOutOrder(String name,String food,String number){UUID takeOutId =UUID.randomUUID();String orderTime =DateFormat.getDateTimeInstance().format(newDate());String exchangeName ="mt_fanout_exchange";String takeOutMes ="美团订单编号："+ takeOutId +" "+ orderTime +" "+ name +" "+ food +" "+ number;String routingKey ="";//将消息放入通道内Object result = rabbitTemplate.convertSendAndReceive(exchangeName, routingKey, takeOutMes);System.out.println("配送中心响应："+result);}

测试类：

@TestvoidmtTakeOutOrder()throwsInterruptedException{
         takeOutOrder.mtTakeOutOrder("小张"+i,"麻辣烫","10086");}

消费者（外卖接单中心）：

正常接单消费者：

@RabbitListener(bindings =@QueueBinding(value =@Queue(value ="mt_queue", autoDelete ="false"),
        exchange =@Exchange(value ="mt_fanout_exchange", type =ExchangeTypes.FANOUT)))@Componentpublicclass mtTakeOutDelivery {@RabbitHandlerpublicStringbuyTrainTickets(String message){System.out.println("正常美团外卖订单已接单："+ message);return"配送中心已接单";}}

加急接单消费者（死信队列内的消息）：

@RabbitListener(bindings =@QueueBinding(value =@Queue(value ="mt_dead_queue", autoDelete ="false"),
        exchange =@Exchange(value ="mt_fanout_dead_exchange", type =ExchangeTypes.FANOUT)))@Componentpublicclass mtDeadTakeOutDelivery {@RabbitHandlerpublicStringbuyTrainTickets(String message){System.out.println("加急饿了么外卖订单已接单："+ message);return"配送中心已接单";}}

消费者运行后，只要监听的两个通道内有消息，就会被消费

六、RabbitMQ的手动ACK

为了确保消息不会丢失，RabbitMQ支持消息应答。消费者发送一个消息应答，告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。

自动应答就是上面的案例，只要被消费者取出，通道内就会删除这个消息，万一这个消息在消费者那边处理异常，因为通道里已经没用这条消息了，就会出现消息丢失。所以在有些场景需要改为手动应答ACK，就是消费者把这条消息确认处理完毕后，再告诉通道删除消息，若异常，这条消息将返回通道内可以重新处理，这就是手动应答。

还是一样，来配置两个通道：

生产者配置：

@ConfigurationpublicclassTestQueueConfig{@BeanpublicFanoutExchangeTestExchange(){returnnewFanoutExchange("test_exchange",true,false);}@BeanpublicFanoutExchangeTestDeadExchange(){returnnewFanoutExchange("test_dead_exchange",true,false);}@BeanpublicQueueTestDeadQueue(){returnnewQueue("test_dead_queue",true,false,false);}@BeanpublicQueueTestQueue(){Map<String,Object> args =newHashMap<>();//20秒钟未消费转到死信队列
        args.put("x-message-ttl",5000);
        args.put("x-dead-letter-exchange","test_dead_exchange");returnnewQueue("test_queue",true,false,false, args);}@BeanpublicBindingTestBinding(){returnBindingBuilder.bind(TestQueue()).to(TestExchange());}@BeanpublicBindingTestDeadBinding(){returnBindingBuilder.bind(TestDeadQueue()).to(TestDeadExchange());}}

生产者业务代码不变，与上个案例一致也行

消费者配置类：

@ConfigurationpublicclassMyselfReceiverConfig{@AutowiredprivateCachingConnectionFactory cachingConnectionFactory;@AutowiredprivateMyselfReceiver myselfReceiver;@AutowiredprivateMyselfDeadReceiver myselfDeadReceiver;@BeanpublicSimpleMessageListenerContainersimpleMessageListenerContainer(){SimpleMessageListenerContainer container =newSimpleMessageListenerContainer(cachingConnectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(10);//手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setQueueNames("test_queue");
        container.setMessageListener(myselfReceiver);return container;}@BeanpublicSimpleMessageListenerContainersimpleMessageListenerContainer_Dead(){SimpleMessageListenerContainer container =newSimpleMessageListenerContainer(cachingConnectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(10);//手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setQueueNames("test_dead_queue");
        container.setMessageListener(myselfDeadReceiver);return container;}}

正常消费者：

@ComponentpublicclassMyselfReceiverimplementsChannelAwareMessageListener{@OverridepublicvoidonMessage(Message message,Channel channel)throwsException{long deliveryTag = message.getMessageProperties().getDeliveryTag();try{byte[] bytes = message.getBody();String mes =newString(bytes);String substring = mes.replace("\\\"","'");System.out.println("正常通道内消息："+substring);//业务主体//若业务处理无异常，则回复通道删除消息
            channel.basicAck(deliveryTag,true);}catch(Exception e){
            channel.basicReject(deliveryTag,true);}}}

死信通道消费者：

@ComponentpublicclassMyselfDeadReceiverimplementsChannelAwareMessageListener{@OverridepublicvoidonMessage(Message message,Channel channel)throwsException{long deliveryTag = message.getMessageProperties().getDeliveryTag();try{byte[] bytes = message.getBody();String mes =newString(bytes);String result = mes.replace("\\\"","'");System.out.println("死信通道内消息："+ result);//业务主体//若业务处理无异常，则回复通道删除消息
            channel.basicAck(deliveryTag,true);}catch(Exception e){//有异常把消息返回通道
            channel.basicReject(deliveryTag,true);}}}

这样就完成了手动ACK，若消费者处理没有异常，将使用

channel.basicAck(deliveryTag, true);

若出现了异常，将使用

channel.basicReject(deliveryTag, true);

，此消息将重新进入通道内，这也确保了未消费成功的消息不会出现丢失的情况。