RabbitMQ的优势

解耦
异步提速
削峰填谷

RabbitMQ的劣势

系统可用性降低
系统复杂度提高
一致性

RabbitMQ的安装（Linux Centos7）

  1. 我们下载最新版的rpm包

image.png

  1. rpm -ivh rabbitmq-server-3.6.8-1.el7.noarch.rpm

然后我们先安装RabbitMQ的依赖erlang，之后在安装RabbitMQ
如果在安装的时候提示缺少socat，使用下面的命令安装

yum install socat logrotate -y

  1. 常用的命令 ```bash RabbitMQ默认的用户名和密码是guest，但是只能从本地localhost登录，所以我们需要添加新的用户

rabbitmqctl add_user user_name user_passwd #设置用户名和密码

rabbitmqctl set_user_tags user_name administrator #设置用户标签

rabbitmqctl set_permissions -p ‘/‘ user_name ‘.’ ‘.’ ‘.’ #设置权限

systemctl status rabbitmq-server #查看RabbitMQ服务状态

rabbitmqctl list_users #列出所有用户

开启管理页面插件，才可以使用web管理端

rabbitmq-plugins enable rabbitmq_management

开启rabbitMQ连接端口

firewall-cmd —zone=public —add-port=15672/tcp —permanent

程序或者其他机器交互使用时：

firewall-cmd —zone=public —add-port=5672/tcp —permanent

firewall-cmd —reload

  1. ```java
  2. #启动流程
  3. rabbitmq-plugins enable rabbitmq_management
  5. systemctl start rabbitmq-server
  8. systemctl status firewalld   #查看防火墙状态
  10. firewall-cmd --list-ports --permanent  #查看永久开放端口
  12. firewall-cmd --add-port=15672/tcp --permanent  #添加永久开放端口
  14. systemctl restart firewalld  #重启防火墙
  16. rabbitmqctl add_user admin
  18. rabbitmqctl set_permissions -p / admin  ".*" ".*" ".*"
  19. rabbitmqctl set_user_tags  admin administrator

image.png

入门

生产者

  1. package com.rabbitmq.producer;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  10. public class Producer_Hello {
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  13.         //创建链接工厂
  14.         ConnectionFactory factory = new ConnectionFactory();
  16.         //设置参数
  17.         factory.setHost("192.168.47.129");
  18.         factory.setPort(5672);
  19.         factory.setVirtualHost("/sky");
  20.         factory.setUsername("admin");
  21.         factory.setPassword("admin");
  22.         //创建链接
  23.         Connection connection = factory.newConnection();
  24.         //创建Channel
  25.         Channel channel = connection.createChannel();
  26.         //创建队列Queue
  27.         channel.queueDeclare("hello",true,false,false,null);
  28.         //发送消息
  29.         String msg = "helloworld";
  30.         channel.basicPublish("","hello",null,msg.getBytes());
  32.         //释放资源
  33.         channel.close();
  34.         connection.close();
  35.     }
  37. }

消费者

  1. package com.rabbitmq.consumer;
  3. import com.rabbitmq.client.*;
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  8. public class Consumer_Hello {
  10.     public static void main(String[] args) throws IOException, TimeoutException {
  11.         //创建链接工厂
  12.         ConnectionFactory factory = new ConnectionFactory();
  14.         //设置参数
  15.         factory.setHost("192.168.47.129");
  16.         factory.setPort(5672);
  17.         factory.setVirtualHost("/sky");
  18.         factory.setUsername("admin");
  19.         factory.setPassword("admin");
  20.         //创建链接
  21.         Connection connection = factory.newConnection();
  22.         //创建Channel
  23.         Channel channel = connection.createChannel();
  24.         //创建队列Queue
  25.         channel.queueDeclare("hello",true,false,false,null);
  26.         DefaultConsumer consumer = new DefaultConsumer(channel){
  27.             @Override
  28.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  29.                 System.out.println(consumerTag);
  30.                 System.out.println(new String(body));
  31.             }
  32.         };
  33.         channel.basicConsume("hello",true,consumer);
  34.     }
  35. }

工作模式WorkQueues

  1. package com.rabbitmq.producer;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  10. public class Producer_work {
  11.     public static void main(String[] args) throws IOException, TimeoutException {
  12.         ConnectionFactory factory = new ConnectionFactory();
  13.         factory.setHost("192.168.47.129");
  14.         factory.setPort(5672);
  15.         factory.setVirtualHost("/sky");
  16.         factory.setUsername("admin");
  17.         factory.setPassword("admin");
  18.         Connection connection = factory.newConnection();
  19.         Channel channel = connection.createChannel();
  20.         channel.queueDeclare("work",true,false,false,null);
  21.         for (int i = 0; i < 10; i++) {
  22.             String msg = i+"helloworld";
  23.             channel.basicPublish("","work",null,msg.getBytes());
  24.         }
  25.         channel.close();
  26.         connection.close();
  27.     }
  28. }
  32. package com.rabbitmq.consumer;
  34. import com.rabbitmq.client.*;
  36. import java.io.IOException;
  37. import java.util.concurrent.TimeoutException;
  39. public class Consumer_work01 {
  40.     public static void main(String[] args) throws IOException, TimeoutException {
  41.         ConnectionFactory factory = new ConnectionFactory();
  42.         factory.setHost("192.168.47.129");
  43.         factory.setPort(5672);
  44.         factory.setVirtualHost("/sky");
  45.         factory.setUsername("admin");
  46.         factory.setPassword("admin");
  47.         Connection connection = factory.newConnection();
  48.         Channel channel = connection.createChannel();
  49.         channel.queueDeclare("work",true,false,false,null);
  50.         DefaultConsumer consumer = new DefaultConsumer(channel){
  51.             @Override
  52.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  53.                 System.out.println(consumerTag);
  54.                 System.out.println(new String(body));
  55.             }
  56.         };
  57.         channel.basicConsume("work",true,consumer);
  58.     }
  59. }

image.png
image.png

订阅模式pub/sub

channel.exchangeDeclare(“pubsub”, BuiltinExchangeType.FANOUT,true,false,false,null); type是fanout

  1. package com.rabbitmq.producer;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. public class Producer_pubsub {
  13.     public static void main(String[] args) throws IOException, TimeoutException {
  14.         ConnectionFactory factory = new ConnectionFactory();
  15.         factory.setHost("192.168.47.129");
  16.         factory.setPort(5672);
  17.         factory.setVirtualHost("/sky");
  18.         factory.setUsername("admin");
  19.         factory.setPassword("admin");
  20.         Connection connection = factory.newConnection();
  21.         Channel channel = connection.createChannel();
  22.         //创建交换机
  23.         channel.exchangeDeclare("pubsub", BuiltinExchangeType.FANOUT,true,false,false,null);
  24.         /**
  25.          * DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type, boolean var3, boolean var4, boolean var5, Map<String, Object> var6)
  26.          *
  27.          * exchange  交换机的名称
  28.          * type  交换机类型
  29.          *     DIRECT("direct"),   定向
  30.          *     FANOUT("fanout"),   扇形
  31.          *     TOPIC("topic"),     通配符方式
  32.          *     HEADERS("headers");  参数匹配
  33.          * 是否持久化
  34.          * 是否自动删除
  35.          * 内部使用 一般false
  36.          * 参数列表
  37.          */
  38.         //创建队列
  39.         channel.queueDeclare("q1",true,false,false,null);
  40.         channel.queueDeclare("q2",true,false,false,null);
  41.         //绑定交换机和队列
  42.         channel.queueBind("q1","pubsub","");
  43.         channel.queueBind("q2","pubsub","");
  44.         channel.basicPublish("pubsub","",null,"1231".getBytes());
  46.         channel.close();
  47.         connection.close();
  48.     }
  49. }
  1. package com.rabbitmq.consumer;
  3. import com.rabbitmq.client.*;
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  8. public class Consumer_pubsub01 {
  9.     public static void main(String[] args) throws IOException, TimeoutException {
  10.         ConnectionFactory factory = new ConnectionFactory();
  11.         factory.setHost("192.168.47.129");
  12.         factory.setPort(5672);
  13.         factory.setVirtualHost("/sky");
  14.         factory.setUsername("admin");
  15.         factory.setPassword("admin");
  16.         Connection connection = factory.newConnection();
  17.         Channel channel = connection.createChannel();
  18.         channel.queueDeclare("q1",true,false,false,null);
  19.         DefaultConsumer consumer = new DefaultConsumer(channel){
  20.             @Override
  21.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  22.                 System.out.println(consumerTag);
  23.                 System.out.println(new String(body));
  24.             }
  25.         };
  26.         channel.basicConsume("q1",true,consumer);
  27.     }
  28. }

路由工作模式

channel.exchangeDeclare(“rote”, BuiltinExchangeType.DIRECT,true,false,false,null); type 是direct

  1. package com.rabbitmq.producer;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. public class Producer_roting {
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  13.         ConnectionFactory factory = new ConnectionFactory();
  14.         factory.setHost("192.168.47.129");
  15.         factory.setPort(5672);
  16.         factory.setVirtualHost("/sky");
  17.         factory.setUsername("admin");
  18.         factory.setPassword("admin");
  19.         Connection connection = factory.newConnection();
  20.         Channel channel = connection.createChannel();
  21.         //创建交换机
  22.         channel.exchangeDeclare("rote", BuiltinExchangeType.DIRECT,true,false,false,null);
  23.         /**
  24.          * DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type, boolean var3, boolean var4, boolean var5, Map<String, Object> var6)
  25.          *
  26.          * exchange  交换机的名称
  27.          * type  交换机类型
  28.          *     DIRECT("direct"),   定向
  29.          *     FANOUT("fanout"),   扇形
  30.          *     TOPIC("topic"),     通配符方式
  31.          *     HEADERS("headers");  参数匹配
  32.          * 是否持久化
  33.          * 是否自动删除
  34.          * 内部使用 一般false
  35.          * 参数列表
  36.          */
  37.         //创建队列
  38.         channel.queueDeclare("q1",true,false,false,null);
  39.         channel.queueDeclare("q2",true,false,false,null);
  40.         //绑定交换机和队列
  41.         channel.queueBind("q1","rote","error");
  42.         channel.queueBind("q2","rote","");
  43.         channel.basicPublish("rote","error",null,"q1接收".getBytes());
  44.         channel.basicPublish("rote","",null,"q2接收".getBytes());
  46.         channel.close();
  47.         connection.close();
  48.     }
  49. }

通配符工作模式

type 是topic

  1. package com.rabbitmq.producer;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. public class Producer_topic {
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  13.         ConnectionFactory factory = new ConnectionFactory();
  14.         factory.setHost("192.168.47.129");
  15.         factory.setPort(5672);
  16.         factory.setVirtualHost("/sky");
  17.         factory.setUsername("admin");
  18.         factory.setPassword("admin");
  19.         Connection connection = factory.newConnection();
  20.         Channel channel = connection.createChannel();
  21.         //创建交换机
  22.         channel.exchangeDeclare("topic", BuiltinExchangeType.TOPIC,true,false,false,null);
  23.         channel.queueDeclare("q1",true,false,false,null);
  24.         channel.queueDeclare("q2",true,false,false,null);
  25.         //绑定交换机和队列
  26.         /**
  27.          * * 表示一个单词
  28.          * # 表示多个单词
  29.          */
  30.         channel.queueBind("q1","topic","*.error");
  31.         channel.queueBind("q1","topic","order.*");
  32.         channel.queueBind("q2","topic","#.#");
  33.         channel.basicPublish("topic","order.aa",null,"q1接收".getBytes());
  34.         channel.basicPublish("topic","",null,"q2接收".getBytes());
  36.         channel.close();
  37.         connection.close();
  38.     }
  39. }

Spring整合RabbitMQ

生产者配置以及代码

  1. #rabbitmq的配置
  2. rabbitmq.host=192.168.47.129
  3. rabbitmq.port=5672
  4. rabbitmq.username=admin
  5. rabbitmq.password=admin
  6. rabbitmq.virtual-host=/sky   #虚拟机名称
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:context="http://www.springframework.org/schema/context"
  5.        xmlns:rabbit="http://www.springframework.org/schema/rabbit"
  6.        xsi:schemaLocation="
  7.        http://www.springframework.org/schema/beans
  8.        http://www.springframework.org/schema/beans/spring-beans.xsd
  9.        http://www.springframework.org/schema/context
  10.        http://www.springframework.org/schema/context/spring-context.xsd
  11.        http://www.springframework.org/schema/rabbit
  12.        http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
  13. ">
  14. <!--    加载配置文件-->
  15.     <context:property-placeholder location="classpath:rabbitmq.properties"/>
  16.     <context:annotation-config/>
  17. <!--    定义连接工厂-->
  18.     <rabbit:connection-factory id="connectionFactory"
  19.                                host="${rabbitmq.host}"
  20.                                port="${rabbitmq.port}"
  21.                                username="${rabbitmq.username}"
  22.                                password="${rabbitmq.password}"
  23.                                virtual-host="${rabbitmq.virtual-host}"
  24.     />
  25. <!--    id是bean的名称  name是队列的名称 auto-declare是自动创建 auto-delete是最后一个客户端断开连接后自动删除队列-->
  26. <!--    定义交换机-->
  27.     <rabbit:admin connection-factory="connectionFactory"/>
  29.     <rabbit:queue id="spring_hello" name="spring_hello" auto-declare="true"/>
  31.     <rabbit:queue id="spring_work1" name="spring_work1" auto-declare="true"/>
  32.     <rabbit:queue id="spring_work2" name="spring_work2" auto-declare="true"/>
  34.     <rabbit:queue id="spring_fanout1" name="spring_fanout1" auto-declare="true"/>
  35.     <rabbit:queue id="spring_fanout2" name="spring_fanout2" auto-declare="true"/>
  37.     <rabbit:fanout-exchange name="spring_exchangefanout" id="spring_exchangefanout" auto-declare="true">
  38.         <rabbit:bindings>
  39.             <rabbit:binding queue="spring_fanout1"/>
  40.             <rabbit:binding queue="spring_fanout2"/>
  41.         </rabbit:bindings>
  42.     </rabbit:fanout-exchange>
  45. <!--    key是指路由key-->
  47.     <rabbit:queue id="spring_route1" name="spring_route1" auto-declare="true"/>
  48.     <rabbit:queue id="spring_route2" name="spring_route2" auto-declare="true"/>
  49.     <rabbit:direct-exchange name="spring_exchangedirect" id="spring_exchangedirect" auto-declare="true">
  50.         <rabbit:bindings>
  51.             <rabbit:binding key="error" queue="spring_route1"/>
  52.             <rabbit:binding queue="spring_route2"/>
  53.         </rabbit:bindings>
  54.     </rabbit:direct-exchange>
  55.     <rabbit:queue id="spring_topic1" name="spring_topic1" auto-declare="true"/>
  56.     <rabbit:queue id="spring_topic2" name="spring_topic2" auto-declare="true"/>
  57. <!--    pattern是指匹配规则-->
  58.     <rabbit:topic-exchange name="spring_exchangetopic" id="spring_exchangetopic" auto-declare="true">
  59.         <rabbit:bindings>
  60.             <rabbit:binding pattern="order.*" queue="spring_topic1"/>
  61.             <rabbit:binding pattern="order.#" queue="spring_topic2"/>
  62.         </rabbit:bindings>
  63.     </rabbit:topic-exchange>
  64.   <!--    rabbitTemplate是用来发送数据的-->
  65.     <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
  66. </beans>
  4. import org.junit.Test;
  5. import org.junit.runner.RunWith;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.test.context.ContextConfiguration;
  9. import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  11. @RunWith(SpringJUnit4ClassRunner.class)
  12. @ContextConfiguration(locations = "classpath:application.xml")
  13. public class Spring_Producer {
  14.     @Autowired
  15.     private RabbitTemplate rabbitTemplate;
  17.     //发送简单工作模式
  18.     @Test
  19.     public void test01(){
  20.         rabbitTemplate.convertAndSend("spring_hello","helloworld");
  21.     }
  23.     //发送广播
  24.     @Test
  25.     public void fanout(){
  26.         rabbitTemplate.convertAndSend("spring_exchangefanout","","广播发送！！");
  27.     }
  28.     @Test
  29.     public void direct(){
  30.         rabbitTemplate.convertAndSend("spring_exchangedirect","error","路由发送！！");
  31.     }
  33.     @Test
  34.     public void topic(){
  35.         rabbitTemplate.convertAndSend("spring_exchangetopic","order.a.a","topic发送！！");
  36.     }
  37. }

消费者配置以及代码

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:context="http://www.springframework.org/schema/context"
  5.        xmlns:rabbit="http://www.springframework.org/schema/rabbit"
  6.        xsi:schemaLocation="
  7.        http://www.springframework.org/schema/beans
  8.        http://www.springframework.org/schema/beans/spring-beans.xsd
  9.        http://www.springframework.org/schema/context
  10.        http://www.springframework.org/schema/context/spring-context.xsd
  11.        http://www.springframework.org/schema/rabbit
  12.        http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
  13. ">
  14. <!--    加载配置文件-->
  15.     <context:property-placeholder location="classpath:rabbitmq.properties"/>
  16.     <context:annotation-config/>
  17. <!--    定义连接工厂-->
  18.     <rabbit:connection-factory id="connectionFactory"
  19.                                host="${rabbitmq.host}"
  20.                                port="${rabbitmq.port}"
  21.                                username="${rabbitmq.username}"
  22.                                password="${rabbitmq.password}"
  23.                                virtual-host="${rabbitmq.virtual-host}"
  24.     />
  25.   <!-- 定义监听器类 -->
  26.     <bean id="SpringQueueHello" class="com.springrabbit.HelloListener"/>
  27.   <!-- 配置监听器容器 ，每一个监听器对应监听的队列 -->
  28.     <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true">
  29.         <rabbit:listener ref="SpringQueueHello" queue-names="spring_hello"/>
  30.     </rabbit:listener-container>
  31. </beans>
  1. #监听类需要实现MessageListener接口并且重写onMessage方法
  2. package com.springrabbit;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.core.MessageListener;
  7. public class HelloListener implements MessageListener {
  8.     @Override
  9.     public void onMessage(Message message) {
  10.         System.out.println(new String(message.getBody()));
  11.     }
  12. }

Springboot整合RabbitMQ

  1. #首先在配置文件中配置RabbitMQ
  3. spring:
  4.   rabbitmq:
  5.     host: 192.168.47.129
  6.     port: 5672
  7.     username: admin
  8.     password: admin
  9.     virtual-host: /sky

生产者

springboot的生产者是一个配置类，在配置类中我们配置各种类型的交换机以及队列，然后定义绑定关系

  1. package com.sbmq.config;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. @Configuration
  9. public class RabbitMQConfig {
  10.     private static final String EXCHANGENAME="boot_exchange";
  12.     @Bean("bootExchange")
  13.     public Exchange bootExchange(){
  14.         return ExchangeBuilder.directExchange(EXCHANGENAME).durable(true).build();
  15.     }
  16.     @Bean("bootQueue")
  17.     public Queue bootQueue(){
  18.         return QueueBuilder.durable("bootqueue").build();
  19.     }
  20.     @Bean
  21.     public Binding bind(@Qualifier("bootQueue") Queue queue,@Qualifier("bootExchange") Exchange exchange){
  22.         return BindingBuilder.bind(queue).to(exchange).with("boot").noargs();
  23.     }
  24. }

消费者

  1. package com.sbmq.config;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  7. @Component
  8. public class Listener {
  9.     @RabbitListener(queues = "bootqueue")
  10.     public void listener(Message message){
  11.         System.out.println(message);
  12.     }
  13. }
  17. 编写一个监听类，使用@RabbitListener注解注明监听的队列，然后方法中接收一个Message类型的参数，里面就是消息

RabbitMQ高级

消息可靠投递

confirm

  1. 1.首先我们需要开启付，默认是关闭的
  2. <rabbit:connection-factory id="connectionFactory"
  3.                                host="${rabbitmq.host}"
  4.                                port="${rabbitmq.port}"
  5.                                username="${rabbitmq.username}"
  6.                                password="${rabbitmq.password}"
  7.                                virtual-host="${rabbitmq.virtual-host}"
  8.                                publisher-confirms="true"
  9.     />
  11. 2.设置回调函数
  12. @Test
  13.     public void test01(){
  14.         rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
  15.         /**
  16.              *
  17.              * @param correlationData  相关配置信息
  18.              * @param b   是否成功
  19.              * @param s   失败的原因
  20.              */
  21.             @Override
  22.             public void confirm(CorrelationData correlationData, boolean b, String s) {
  23.                 System.out.println("huidioa");
  24.             }
  25.         });
  26.         rabbitTemplate.convertAndSend("spring_hello","helloworld");
  27.     }

return

  1. @Test
  2.     public void test01(){
  3.         /**
  4.          * 1:开启服务
  5.          * 2：设置回调函数
  6.          * 3：设置模式
  7.          *      默认失败丢弃消息
  8.          *      失败返回消息
  9.          */
  10.         rabbitTemplate.setMandatory(true);
  11.         rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
  12.             /**
  13.              *
  14.              * @param message  消息数据
  15.              * @param i  错误码
  16.              * @param s   错误信息
  17.              * @param s1  交换机
  18.              * @param s2  路由key
  19.              */
  20.             @Override
  21.             public void returnedMessage(Message message, int i, String s, String s1, String s2) {
  22.                 System.out.println("return");
  23.             }
  24.         });
  25.         rabbitTemplate.convertAndSend("spring_hello12","helloworld");
  26.     }

ACK 消息可靠性接收

  1. <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true" acknowledge="manual">
  2.         <rabbit:listener ref="SpringQueueHello" queue-names="spring_hello"/>
  3. </rabbit:listener-container>
  6. acknowledge="manual"
  8. manual  手动
  9. auto    根据异常
  10. none    自动接收
  1. package com.springrabbit;
  3. import com.rabbitmq.client.Channel;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.core.MessageListener;
  6. import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
  8. import java.io.IOException;
  10. public class HelloListener implements ChannelAwareMessageListener {
  12.     @Override
  13.     public void onMessage(Message message, Channel channel) throws Exception {
  14.         long deliveryTag = message.getMessageProperties().getDeliveryTag();
  15.         try {
  16.             Thread.sleep(1000);
  17.             System.out.println(message);
  18.             /**
  19.              * 当前标签
  20.              * true签收所有消息
  21.              */
  22.             channel.basicAck(deliveryTag,true);
  23.         } catch (Exception e) {
  24.             channel.basicNack(deliveryTag,true,true);//第三个参数true是返回队列重新发送
  25.         }
  26.     }
  27. }

消费端限流（削峰填谷）

  1. 只需要设置监听容器新增加一个属性
  2. <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true"
  3.     acknowledge="manual" prefetch="1">
  4.         <rabbit:listener ref="SpringQueueHello" queue-names="spring_hello"/>
  5. </rabbit:listener-container>
  7. ## 1:需要设置手动签收消息
  8. ## 2:prefetch 里面的数表示一次接收几个消息，签收后才继续接收消息

TTL（Time To Live）存活/过期

  1. ##1:对队列设置过期时间
  2. 使用参数x-message-ttl,单位ms
  3. <rabbit:queue id="spring_topic1" name="spring_topic1" auto-declare="true">
  4.         <rabbit:queue-arguments>
  5.             <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
  6.         </rabbit:queue-arguments>
  7. </rabbit:queue>
  10. ##2：对消息单独设置过期时间
  11. @Test
  12. public void test01(){
  13.   MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
  14.     @Override
  15.     public Message postProcessMessage(Message message) throws AmqpException {
  16.     message.getMessageProperties().setExpiration("50000");//过期时间
  17.     return message;
  18.     }
  19.   };
  20.   rabbitTemplate.convertAndSend("spring_hello","","helloworld",messagePostProcessor);
  21. }

死信队列

  1. # 生产者配置
  2. <!--    死信队列
  3. 1:正常的队列和交换机
  4. 2：死信队列和交换机
  5. -->
  6. <!--    正常队列-->
  7.     <rabbit:queue id="dead_queue" name="dead_queue" auto-declare="true">
  8.         <rabbit:queue-arguments>
  9. <!--            绑定死信交换机-->
  10.             <entry key="x-dead-letter-exchange" value="deaded_exchange"/>
  11. <!--            设置发送消息的路由key-->
  12.             <entry key="x-dead-letter-routing-key" value="deaded.test"/>
  13. <!--         设置正常队列消息死亡时间以及长度-->
  14.             <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
  15.             <entry key="x-max-length" value="10" value-type="java.lang.Integer"/>
  16.         </rabbit:queue-arguments>
  17.     </rabbit:queue>
  19.     <rabbit:topic-exchange name="dead_exchange" id="dead_exchange" auto-declare="true">
  20.         <rabbit:bindings>
  21.             <rabbit:binding pattern="dead.*" queue="dead_queue"/>
  22.         </rabbit:bindings>
  23.     </rabbit:topic-exchange>
  24. <!--    死信队列-->
  25.     <rabbit:queue id="deaded_queue" name="deaded_queue" auto-declare="true"/>
  26.     <rabbit:topic-exchange name="deaded_exchange">
  27.         <rabbit:bindings>
  28.             <rabbit:binding pattern="deaded.*" queue="deaded_queue"/>
  29.         </rabbit:bindings>
  30.     </rabbit:topic-exchange>
  1. # 消费者配置以及代码
  3.     <bean id="DeadQueue" class="com.springrabbit.DeadListener"/>
  4.     <bean id="DeadedQueue" class="com.springrabbit.DeadedListener"/>
  5.     <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true" acknowledge="manual" prefetch="1">
  6.         <rabbit:listener ref="SpringQueueHello" queue-names="spring_hello"/>
  7.         <rabbit:listener ref="DeadQueue" queue-names="dead_queue"/>
  8.         <rabbit:listener ref="DeadedQueue" queue-names="deaded_queue"/>
  9.     </rabbit:listener-container>
  12. package com.springrabbit;
  14. import com.rabbitmq.client.Channel;
  15. import org.springframework.amqp.core.Message;
  16. import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
  18. public class DeadedListener implements ChannelAwareMessageListener {
  19.     @Override
  20.     public void onMessage(Message message, Channel channel) throws Exception {
  21.         //接收死信队列的消息
  22.         long deliveryTag = message.getMessageProperties().getDeliveryTag();
  23.         System.out.println(new String(message.getBody())+"死信");
  24.         channel.basicAck(deliveryTag,true);
  25.     }
  26. }
  30. package com.springrabbit;
  32. import com.rabbitmq.client.Channel;
  33. import org.springframework.amqp.core.Message;
  34. import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
  36. public class DeadListener implements ChannelAwareMessageListener {
  37.     @Override
  38.     public void onMessage(Message message, Channel channel) throws Exception {
  39.         long deliveryTag = message.getMessageProperties().getDeliveryTag();
  40.         try {
  41.             System.out.println(new String(message.getBody()));
  42.             int i = 1/0;
  43.             channel.basicAck(deliveryTag,true);
  44.         } catch (Exception e) {
  45.             channel.basicNack(deliveryTag,true,false);
  46.         }
  47.     }
  48. }

延时队列

日志与监控

消息追踪

  1. # 开启 ，默认开启/虚拟机
  2. [root@localhost rabbitmq]# rabbitmqctl trace_on
  3. # 关闭
  4. [root@localhost rabbitmq]# rabbitmqctl trace_off

更强大的插件 rabbitmq_tracing 

  1. # 开启
  2. rabbitmq-plugins enable rabbitmq_tracing
  4. # 查看所有插件
  5. rabbitmq-plugins list

应用问题

消息补偿

幂等性保障

集群搭建

镜像队列