定制小程序开发费用学习RabbitMQ这篇就够了快速入门上手(超详细)

  • 💂 个人主页: 
  • 🤟 版权: 本文由【陶然同学】原创、在CSDN首发、定制小程序开发费用需要转载请联系博主
  • 💬 定制小程序开发费用如果文章对你有帮助、欢迎关注、点赞、收藏(一键三连)定制小程序开发费用和订阅专栏哦
  • 💅 定制小程序开发费用想寻找共同成长的小伙伴，请点击【】

目录

1.MQ

1.1什么是MQ

消息队列，即MQ，Message Queue。

  1. 定制小程序开发费用消息队列是典型的：生产者、定制小程序开发费用消费者模型。
    1. 定制小程序开发费用生产者不断向消息队列定制小程序开发费用中生产消息，
    2. 定制小程序开发费用消费者不断的从队列中获取消息。
    3. 定制小程序开发费用因为消息的生产和消费定制小程序开发费用都是异步的

定制小程序开发费用而且只关心消息的发送和接收，定制小程序开发费用没有业务逻辑的侵入，定制小程序开发费用这样就实现了生产者和定制小程序开发费用消费者的解耦。

1.2实现方式：AMQP、JMS

MQ定制小程序开发费用是消息通信的模型，定制小程序开发费用并不是具体实现。现在实现MQ定制小程序开发费用的有两种主流方式：AMQP、JMS。

  1. 定制小程序开发费用两者间的区别和联系：
  1. JMS定制小程序开发费用是定义了统一的接口，定制小程序开发费用来对消息操作进行统一；AMQP定制小程序开发费用是通过规定协议来统一数据交互的格式
  2. JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
  3. JMS规定了两种消息模型；而AMQP的消息模型更加丰富

1.3常见MQ产品

  1. ActiveMQ：基于JMS
  2. RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
  3. RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
  4. Kafka：分布式消息系统，高吞吐量

2.

2.1RabbitMQ概述

2.1.1什么是Rabbit

 MQ全称为Message Queue，即消息队列， RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。

RabbitMQ官方地址：

快速入门：

2.1.2RabbitMQ的工作原理

组成部分说明如下

  1. Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。
  2. Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
  3. Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
  4. Producer：消息生产者，即生产方客户端，生产方客户端将消费发送到MQ。
  5. Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。

2.1.3消息流程

  1. 发送消息流程

1、生产者和Broker建立TCP连接。

2、生产者和Broker建立通道。

3、生产者通过通道消息发送Broker，由Exchange将消息进行转发。

4、Exchange将消息转发到指定的Queue（队列）

  1. 接收消息流程

1、消费者和Broker建立TCP连接

2、消费者和Broker建立通道

3、消费者监听指定的Queue（队列）

4、当有消息到达Queue时Broker默认将消息推送给消费者。

5、消费者接收到消息。

2.2下载和安装

2.2.1下载

  1. 官网下载地址：
  2. 目前最新版本是：3.8.3

2.2.2安装

  1. 安装erlang

  1. 安装RabbitMQ

2.2.3运行

  • 步骤一：打开控制台

  • 步骤二：激活插件

rabbitmq-plugins.bat enable rabbitmq_management

  • 步骤三：重启

net stop rabbitmq && net start rabbitmq

2.2.4访问

  • 步骤一：输入

  • 步骤二：默认账号登录：guest/ guest

2.3入门案例

2.3.1环境搭建

  • 构建环境：test_rabbitmq

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.springframework.boot</groupId>
  4.         <artifactId>spring-boot-starter-amqp</artifactId>
  5.     </dependency>
  6. </dependencies>

2.3.2生产者

  1. ackage com.czxy.xuecheng.rabbitmq;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  9.  
  10. /**
  11.  * Created by liangtong.
  12.  */
  13. public class Producer01 {
  14.     //队列名称
  15.     private static final String QUEUE = "helloworld";
  16.     
  17.     public static void main(String[] args) throws IOException, TimeoutException {
  18.         Connection connection = null;
  19.         Channel channel = null;
  20.         try {
  21.             ConnectionFactory factory = new ConnectionFactory();
  22.             factory.setHost("localhost");
  23.             factory.setPort(5672);
  24.             factory.setUsername("guest");
  25.             factory.setPassword("guest");
  26.             factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  27.             // 创建与RabbitMQ服务的TCP连接
  28.             connection = factory.newConnection();
  29.             //创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
  30.             channel = connection.createChannel();
  31.             /**
  32.              * 声明队列，如果Rabbit中没有此队列将自动创建
  33.              * param1:队列名称
  34.              * param2:是否持久化
  35.              * param3:队列是否独占此连接
  36.              * param4:队列不再使用时是否自动删除此队列
  37.              * param5:队列参数
  38.              */
  39.             channel.queueDeclare(QUEUE, true, false, false, null);
  40.             String message = "helloworld小明"+System.currentTimeMillis();
  41.             /**
  42.              * 消息发布方法
  43.              * param1：Exchange的名称，如果没有指定，则使用Default Exchange
  44.              * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
  45.              * param3:消息包含的属性
  46.              * param4：消息体
  47.              */
  48.             /**
  49.              * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显
  50.              示绑定或解除绑定
  51.              * 默认的交换机，routingKey等于队列名称
  52.              */
  53.             channel.basicPublish("", QUEUE, null, message.getBytes());
  54.             System.out.println("Send Message is:'" + message + "'");
  55.         } catch(Exception ex) {
  56.             ex.printStackTrace();
  57.         } finally {
  58.             if(channel != null) {
  59.                 channel.close();
  60.             }
  61.             if(connection != null) {
  62.                 connection.close();
  63.             }
  64.         }
  65.     }
  66.  
  67. }

2.3.3消费者

  1. package com.czxy.xuecheng.rabbitmq;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. /**
  9.  * Created by liangtong.
  10.  */
  11. public class Consumer01 {
  12.  
  13.     private static final String QUEUE = "helloworld";
  14.     public static void main(String[] args) throws IOException, TimeoutException {
  15.         ConnectionFactory factory = new ConnectionFactory();
  16.         //设置MabbitMQ所在服务器的ip和端口
  17.         factory.setHost("127.0.0.1");
  18.         factory.setPort(5672);
  19.         Connection connection = factory.newConnection();
  20.         Channel channel = connection.createChannel();
  21.         //声明队列
  22.         channel.queueDeclare(QUEUE, true, false, false, null);
  23.         //定义消费方法
  24.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  25.             /**
  26.              * 消费者接收消息调用此方法
  27.              * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
  28.              * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
  29.              * @param properties
  30.              * @param body
  31.              * @throws IOException
  32.              */
  33.             @Override
  34.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  35.                 //交换机
  36.                 String exchange = envelope.getExchange();
  37.                 //路由key
  38.                 String routingKey = envelope.getRoutingKey();
  39.                 //消息id
  40.                 long deliveryTag = envelope.getDeliveryTag();
  41.                 //消息内容
  42.                 String msg = new String(body,"utf-8");
  43.                 System.out.println("receive message.." + msg);
  44.             }
  45.         };
  46.         /**
  47.         * 监听队列String queue, boolean autoAck,Consumer callback
  48.         * 参数明细
  49.         * 1、队列名称
  50.         * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动回复
  51.         * 3、消费消息的方法，消费者接收到消息后调用此方法
  52.         */
  53.         channel.basicConsume(QUEUE, true, consumer);
  54.     }
  55. }

2.4工作模式

RabbitMQ有以下几种工作模式 ：

1、Work queues

2、Publish/Subscribe：发布订阅模式

3、Routing：路由模式

4、Topics

5、Header

6、RPC

2.4.1Word queues

work queues与入门程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息。

  1. * 测试：

1、使用入门程序，启动多个消费者。

2、生产者发送多个消息。

  1. 结果：

1、一条消息只会被一个消费者接收；

2、rabbit采用轮询的方式将消息是平均发送给消费者的；

3、消费者在处理完某条消息后，才会收到下一条消息。

2.4.2Publish/Subscribe

发布订阅模式：

1、每个消费者监听自己的队列。

2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息

  1. 需求：用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法 。
  2. 1）生产者

声明inform_Exchange_fanout交换机。

声明两个队列并且绑定到此交换机，绑定时不需要指定routingkey

发送消息时不需要指定routingkey

  1. package com.czxy.xuecheng.rabbitmq.demo02;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. import java.io.IOException;
  9.  
  10. /**
  11.  * Created by liangtong.
  12.  */
  13. public class Producer02_publish {
  14.  
  15.     //队列名称
  16.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
  17.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
  18.     private static final String EXCHANGE_FANOUT_INFORM = "inform_exchange_fanout";
  19.  
  20.     public static void main(String[] args) {
  21.         Connection connection = null;
  22.         Channel channel = null;
  23.         try {
  24.             //创建一个与MQ的连接
  25.             ConnectionFactory factory = new ConnectionFactory();
  26.             factory.setHost("127.0.0.1");
  27.             factory.setPort(5672);
  28.             factory.setUsername("guest");
  29.             factory.setPassword("guest");
  30.             factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  31.  
  32.             //创建一个连接
  33.             connection = factory.newConnection();
  34.             //创建与交换机的通道，每个通道代表一个会话
  35.             channel = connection.createChannel();
  36.             //声明交换机 String exchange, BuiltinExchangeType type
  37.             /**
  38.              * 参数明细
  39.              * 1、交换机名称
  40.              * 2、交换机类型，fanout、topic、direct、headers
  41.              *  fanout：广播模式需要生产者消费者绑定相同的Exchange
  42.              *  topic：支持模糊匹配的广播模式以点分隔，*表示一个单词，#表示任意数量（零个或多个）单词
  43.              *  direct ：需要生产者和消费者绑定相同的Exchange和routing key
  44.              *  headers：根据生产者和消费者的header中信息进行匹配性能较差 ，x-match [all 匹配所有/any 任意一个]。
  45.              */
  46.             channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
  47.             //声明队列
  48.             // (String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
  49.             /**
  50.              * 参数明细：
  51.              * 1、队列名称
  52.              * 2、是否持久化
  53.              * 3、是否独占此队列
  54.              * 4、队列不用是否自动删除
  55.              * 5、参数
  56.              */
  57.             channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
  58.             channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
  59.             //交换机和队列绑定String queue, String exchange, String routingKey
  60.             /**
  61.              * 参数明细
  62.              * 1、队列名称
  63.              * 2、交换机名称
  64.              * 3、路由key
  65.              */
  66.             channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");
  67.             channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_FANOUT_INFORM, "");
  68.             //发送消息
  69.             for (int i = 0; i < 10; i++) {
  70.                 String message = "inform to user" + i;
  71.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
  72.                 /**
  73.                  * 参数明细
  74.                  * 1、交换机名称，不指令使用默认交换机名称 Default Exchange
  75.                  * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消息将发到此队列
  76.                  * 3、消息属性
  77.                  * 4、消息内容
  78.                  */
  79.                 channel.basicPublish(EXCHANGE_FANOUT_INFORM, "", null, message.getBytes());
  80.                 System.out.println("Send Message is:'" + message + "'");
  81.             }
  82.  
  83.         } catch (Exception e) {
  84.             e.printStackTrace();
  85.         } finally {
  86.             if (channel != null) {
  87.                 try {
  88.                     channel.close();
  89.                 } catch (Exception e) {
  90.                     e.printStackTrace();
  91.                 }
  92.             }
  93.             if (connection != null) {
  94.                 try {
  95.                     connection.close();
  96.                 } catch (IOException e) {
  97.                     e.printStackTrace();
  98.                 }
  99.             }
  100.  
  101.         }
  102.     }
  103. }
  1. 2）邮件发送消费者
  1. package com.czxy.xuecheng.rabbitmq.demo02;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  * Created by liangtong.
  9.  */
  10. public class Consumer02_subscribe_email {
  11.     //队列名称
  12.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
  13.     private static final String EXCHANGE_FANOUT_INFORM = "inform_exchange_fanout";
  14.  
  15.     public static void main(String[] args) throws Exception {
  16.         //创建一个与MQ的连接
  17.         ConnectionFactory factory = new ConnectionFactory();
  18.         factory.setHost("127.0.0.1");
  19.         factory.setPort(5672);
  20.         factory.setUsername("guest");
  21.         factory.setPassword("guest");
  22.         factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  23.  
  24.         //创建一个连接
  25.         Connection connection = factory.newConnection();
  26.         //创建与交换机的通道，每个通道代表一个会话
  27.         Channel channel = connection.createChannel();
  28.         //声明交换机 String exchange, BuiltinExchangeType type
  29.         /**
  30.          * 参数明细
  31.          * 1、交换机名称
  32.          * 2、交换机类型，fanout、topic、direct、headers
  33.          */
  34.         channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
  35.         //声明队列
  36.         //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
  37.         /**
  38.          * 参数明细：
  39.          * 1、队列名称
  40.          * 2、是否持久化
  41.          * 3、是否独占此队列
  42.          * 4、队列不用是否自动删除
  43.          * 5、参数
  44.          */
  45.         channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
  46.         //交换机和队列绑定String queue, String exchange, String routingKey
  47.         /**
  48.          * 参数明细
  49.          * 1、队列名称
  50.          * 2、交换机名称
  51.          * 3、路由key
  52.          */
  53.         channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");
  54.         //定义消费方法
  55.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
  56.             @Override
  57.             public void handleDelivery(String consumerTag, Envelope envelope,
  58.                                        AMQP.BasicProperties properties, byte[] body) throws IOException {
  59.                 long deliveryTag = envelope.getDeliveryTag();
  60.                 String exchange = envelope.getExchange();
  61.                 //消息内容
  62.                 String message = new String(body, "UTF-8");
  63.                 System.out.println(message);
  64.             }
  65.         };
  66.         /**
  67.          * 监听队列String queue, boolean autoAck,Consumer callback
  68.          * 参数明细
  69.          * 1、队列名称
  70.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
  71.          为false则需要手动回复
  72.          * 3、消费消息的方法，消费者接收到消息后调用此方法
  73.          */
  74.         channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer);
  75.  
  76.  
  77.     }
  78. }
  1. 3）短信发送消费者
  1. package com.czxy.xuecheng.rabbitmq.demo02;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  * Created by liangtong.
  9.  */
  10. public class Consumer02_subscribe_sms {
  11.     //队列名称
  12.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
  13.     private static final String EXCHANGE_FANOUT_INFORM = "inform_exchange_fanout";
  14.  
  15.     public static void main(String[] args) throws Exception {
  16.         //创建一个与MQ的连接
  17.         ConnectionFactory factory = new ConnectionFactory();
  18.         factory.setHost("127.0.0.1");
  19.         factory.setPort(5672);
  20.         factory.setUsername("guest");
  21.         factory.setPassword("guest");
  22.         //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  23.         factory.setVirtualHost("/");
  24.  
  25.         //创建一个连接
  26.         Connection connection = factory.newConnection();
  27.         //创建与交换机的通道，每个通道代表一个会话
  28.         Channel channel = connection.createChannel();
  29.         //声明交换机 String exchange, BuiltinExchangeType type
  30.         /**
  31.          * 参数明细
  32.          * 1、交换机名称
  33.          * 2、交换机类型，fanout、topic、direct、headers
  34.          */
  35.         channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
  36.         //声明队列
  37.         //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
  38.         /**
  39.          * 参数明细：
  40.          * 1、队列名称
  41.          * 2、是否持久化
  42.          * 3、是否独占此队列
  43.          * 4、队列不用是否自动删除
  44.          * 5、参数
  45.          */
  46.         channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
  47.         //交换机和队列绑定String queue, String exchange, String routingKey
  48.         /**
  49.          * 参数明细
  50.          * 1、队列名称
  51.          * 2、交换机名称
  52.          * 3、路由key
  53.          */
  54.         channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_FANOUT_INFORM, "");
  55.         //定义消费方法
  56.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
  57.             @Override
  58.             public void handleDelivery(String consumerTag, Envelope envelope,
  59.                                        AMQP.BasicProperties properties, byte[] body) throws IOException {
  60.                 long deliveryTag = envelope.getDeliveryTag();
  61.                 String exchange = envelope.getExchange();
  62.                 //消息内容
  63.                 String message = new String(body, "UTF-8");
  64.                 System.out.println(message);
  65.             }
  66.         };
  67.         /**
  68.          * 监听队列String queue, boolean autoAck,Consumer callback
  69.          * 参数明细
  70.          * 1、队列名称
  71.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
  72.          为false则需要手动回复
  73.          * 3、消费消息的方法，消费者接收到消息后调用此方法
  74.          */
  75.         channel.basicConsume(QUEUE_INFORM_SMS, true, defaultConsumer);
  76.  
  77.  
  78.     }
  79. }

2.4.3Routing

路由模式：

1、每个消费者监听自己的队列，并且设置routingkey。

2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列。

  1. 1）生产者

声明inform_exchange_routing交换机。

声明两个队列并且绑定到此交换机，绑定时需要指定routingkey

发送消息时需要指定routingkey

  1. package com.czxy.xuecheng.rabbitmq.demo03;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. import java.io.IOException;
  9.  
  10. /**
  11.  * Created by liangtong.
  12.  */
  13. public class Producer03_routing {
  14.     //队列名称
  15.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
  16.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
  17.     private static final String EXCHANGE_ROUTING_INFORM = "inform_exchange_routing";
  18.  
  19.     public static void main(String[] args) {
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             //创建一个与MQ的连接
  24.             ConnectionFactory factory = new ConnectionFactory();
  25.             factory.setHost("127.0.0.1");
  26.             factory.setPort(5672);
  27.             factory.setUsername("guest");
  28.             factory.setPassword("guest");
  29.             //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  30.             factory.setVirtualHost("/");
  31.  
  32.             //创建一个连接
  33.             connection = factory.newConnection();
  34.             //创建与交换机的通道，每个通道代表一个会话
  35.             channel = connection.createChannel();
  36.             //声明交换机 String exchange, BuiltinExchangeType type
  37.             /**
  38.              * 参数明细
  39.              * 1、交换机名称
  40.              * 2、交换机类型，fanout、topic、direct、headers
  41.              */
  42.             channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
  43.             //声明队列
  44.             //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map < String, Object > arguments)
  45.             /**
  46.              * 参数明细：
  47.              * 1、队列名称
  48.              * 2、是否持久化
  49.              * 3、是否独占此队列
  50.              * 4、队列不用是否自动删除
  51.              * 5、参数
  52.              */
  53.             channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
  54.             channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
  55.             //交换机和队列绑定String queue, String exchange, String routingKey
  56.             /**
  57.              * 参数明细
  58.              * 1、队列名称
  59.              * 2、交换机名称
  60.              * 3、路由key
  61.              */
  62.             channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL);
  63.             channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS);
  64.             //发送邮件消息
  65.             for (int i = 0; i < 10; i++) {
  66.                 String message = "email inform to user" + i;
  67.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body
  68.                 /**
  69.                  * 参数明细
  70.                  * 1、交换机名称，不指令使用默认交换机名称 Default Exchange
  71.                  * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消
  72.                  息将发到此队列
  73.                  * 3、消息属性
  74.                  * 4、消息内容
  75.                  */
  76.                 channel.basicPublish(EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL, null, message.getBytes());
  77.                 System.out.println("Send Message is:'" + message + "'");
  78.             }
  79.             //发送短信消息
  80.             for (int i = 0; i < 10; i++) {
  81.                 String message = "sms inform to user" + i;
  82.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body
  83.                 channel.basicPublish(EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS, null, message.getBytes());
  84.                 System.out.println("Send Message is:'" + message + "'");
  85.             }
  86.  
  87.         } catch (Exception e) {
  88.             e.printStackTrace();
  89.         }  finally {
  90.             if (channel != null) {
  91.                 try {
  92.                     channel.close();
  93.                 } catch (Exception e) {
  94.                     e.printStackTrace();
  95.                 }
  96.             }
  97.             if (connection != null) {
  98.                 try {
  99.                     connection.close();
  100.                 } catch (IOException e) {
  101.                     e.printStackTrace();
  102.                 }
  103.             }
  104.  
  105.         }
  106.  
  107.  
  108.     }
  109. }
  1. 2）邮件发送消费者
  1. package com.czxy.xuecheng.rabbitmq.demo03;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  * Created by liangtong.
  9.  */
  10. public class Consumer03_routing_email {
  11.     //队列名称
  12.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
  13.     private static final String EXCHANGE_ROUTING_INFORM = "inform_exchange_routing";
  14.  
  15.     public static void main(String[] args) throws Exception {
  16.         //创建一个与MQ的连接
  17.         ConnectionFactory factory = new ConnectionFactory();
  18.         factory.setHost("127.0.0.1");
  19.         factory.setPort(5672);
  20.         factory.setUsername("guest");
  21.         factory.setPassword("guest");
  22.         //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  23.         factory.setVirtualHost("/");
  24.  
  25.         //创建一个连接
  26.         Connection connection = factory.newConnection();
  27.         //创建与交换机的通道，每个通道代表一个会话
  28.         Channel channel = connection.createChannel();
  29.         //声明交换机 String exchange, BuiltinExchangeType type
  30.         /**
  31.          * 参数明细
  32.          * 1、交换机名称
  33.          * 2、交换机类型，fanout、topic、direct、headers
  34.          */
  35.         channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
  36.         //声明队列
  37.         //channel.queueDeclare(String queue, boolean durable, boolean exclusive, booleanautoDelete, Map<String, Object> arguments)
  38.         /**
  39.          * 参数明细：
  40.          * 1、队列名称
  41.          * 2、是否持久化
  42.          * 3、是否独占此队列
  43.          * 4、队列不用是否自动删除
  44.          * 5、参数
  45.          */
  46.         channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
  47.         //交换机和队列绑定String queue, String exchange, String routingKey
  48.         /**
  49.          * 参数明细
  50.          * 1、队列名称
  51.          * 2、交换机名称
  52.          * 3、路由key
  53.          */
  54.         channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL);
  55.         //定义消费方法
  56.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
  57.             @Override
  58.             public void handleDelivery(String consumerTag, Envelope envelope,
  59.                                        AMQP.BasicProperties properties, byte[] body) throws IOException {
  60.                 long deliveryTag = envelope.getDeliveryTag();
  61.                 String exchange = envelope.getExchange();
  62.                 //消息内容
  63.                 String message = new String(body, "UTF-8");
  64.                 System.out.println(message);
  65.             }
  66.         };
  67.         /**
  68.          * 监听队列String queue, boolean autoAck,Consumer callback
  69.          * 参数明细
  70.          * 1、队列名称
  71.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
  72.          为false则需要手动回复
  73.          * 3、消费消息的方法，消费者接收到消息后调用此方法
  74.          */
  75.         channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer);
  76.  
  77.  
  78.     }
  79. }
  1. 3）短信发送消费者

参考邮件发送消费者的代码流程，编写短信通知的代码。

  1. package com.czxy.xuecheng.rabbitmq.demo03;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  * Created by liangtong.
  9.  */
  10. public class Consumer03_routing_sms {
  11.     //队列名称
  12.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
  13.     private static final String EXCHANGE_ROUTING_INFORM = "inform_exchange_routing";
  14.  
  15.     public static void main(String[] args) throws Exception {
  16.         //创建一个与MQ的连接
  17.         ConnectionFactory factory = new ConnectionFactory();
  18.         factory.setHost("127.0.0.1");
  19.         factory.setPort(5672);
  20.         factory.setUsername("guest");
  21.         factory.setPassword("guest");
  22.         //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  23.         factory.setVirtualHost("/");
  24.  
  25.         //创建一个连接
  26.         Connection connection = factory.newConnection();
  27.         //创建与交换机的通道，每个通道代表一个会话
  28.         Channel channel = connection.createChannel();
  29.         //声明交换机 String exchange, BuiltinExchangeType type
  30.         /**
  31.          * 参数明细
  32.          * 1、交换机名称
  33.          * 2、交换机类型，fanout、topic、direct、headers
  34.          */
  35.         channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
  36.         //声明队列
  37.         //channel.queueDeclare(String queue, boolean durable, boolean exclusive, booleanautoDelete, Map<String, Object> arguments)
  38.         /**
  39.          * 参数明细：
  40.          * 1、队列名称
  41.          * 2、是否持久化
  42.          * 3、是否独占此队列
  43.          * 4、队列不用是否自动删除
  44.          * 5、参数
  45.          */
  46.         channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
  47.         //交换机和队列绑定String queue, String exchange, String routingKey
  48.         /**
  49.          * 参数明细
  50.          * 1、队列名称
  51.          * 2、交换机名称
  52.          * 3、路由key
  53.          */
  54.         channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS);
  55.         //定义消费方法
  56.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
  57.             @Override
  58.             public void handleDelivery(String consumerTag, Envelope envelope,
  59.                                        AMQP.BasicProperties properties, byte[] body) throws IOException {
  60.                 long deliveryTag = envelope.getDeliveryTag();
  61.                 String exchange = envelope.getExchange();
  62.                 //消息内容
  63.                 String message = new String(body, "UTF-8");
  64.                 System.out.println(message);
  65.             }
  66.         };
  67.         /**
  68.          * 监听队列String queue, boolean autoAck,Consumer callback
  69.          * 参数明细
  70.          * 1、队列名称
  71.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
  72.          为false则需要手动回复
  73.          * 3、消费消息的方法，消费者接收到消息后调用此方法
  74.          */
  75.         channel.basicConsume(QUEUE_INFORM_SMS, true, defaultConsumer);
  76.  
  77.     }
  78. }

2.4.4Topic

  1. 主题模式

1、每个消费者监听自己的队列，并且设置带统配符的routingkey。

2、生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。

  1. 需求：根据用户的通知设置去通知用户，设置接收Email的用户只接收Email，设置接收sms的用户只接收sms，设置两种通知类型都接收的则两种通知都有效。

  1. 1）生产者

队列绑定交换机指定通配符：

统配符规则：

中间以“.”分隔。

符号#可以匹配多个词，符号*可以匹配一个词语。

  1. package com.czxy.xuecheng.rabbitmq.demo04;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. import java.io.IOException;
  9.  
  10. /**
  11.  * Created by liangtong.
  12.  */
  13. public class Producer04_topic {
  14.     //队列名称
  15.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
  16.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
  17.     private static final String EXCHANGE_TOPIC_INFORM = "inform_exchange_topic";
  18.  
  19.     public static void main(String[] args) {
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             //创建一个与MQ的连接
  24.             ConnectionFactory factory = new ConnectionFactory();
  25.             factory.setHost("127.0.0.1");
  26.             factory.setPort(5672);
  27.             factory.setUsername("guest");
  28.             factory.setPassword("guest");
  29.             //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  30.             factory.setVirtualHost("/");
  31.  
  32.             //创建一个连接
  33.             connection = factory.newConnection();
  34.             //创建与交换机的通道，每个通道代表一个会话
  35.             channel = connection.createChannel();
  36.             //声明交换机 String exchange, BuiltinExchangeType type
  37.             /**
  38.              * 参数明细
  39.              * 1、交换机名称
  40.              * 2、交换机类型，fanout、topic、direct、headers
  41.              */
  42.             channel.exchangeDeclare(EXCHANGE_TOPIC_INFORM, BuiltinExchangeType.TOPIC);
  43.             //声明队列
  44.             //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map < String, Object > arguments)
  45.             /**
  46.              * 参数明细：
  47.              * 1、队列名称
  48.              * 2、是否持久化
  49.              * 3、是否独占此队列
  50.              * 4、队列不用是否自动删除
  51.              * 5、参数
  52.              */
  53.             channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
  54.             channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
  55.             //交换机和队列绑定String queue, String exchange, String routingKey
  56.             /**
  57.              * 参数明细
  58.              * 1、队列名称
  59.              * 2、交换机名称
  60.              * 3、路由key
  61.              */
  62.             channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPIC_INFORM, "inform.#.email.#");
  63.             channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_TOPIC_INFORM, "inform.#.sms.#");
  64.             //发送邮件消息
  65.             for (int i = 0; i < 10; i++) {
  66.                 String message = "email inform to user" + i;
  67.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body
  68.                 /**
  69.                  * 参数明细
  70.                  * 1、交换机名称，不指令使用默认交换机名称 Default Exchange
  71.                  * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消
  72.                  息将发到此队列
  73.                  * 3、消息属性
  74.                  * 4、消息内容
  75.                  */
  76.                 channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.email", null, message.getBytes());
  77.                 System.out.println("Send Message is:'" + message + "'");
  78.             }
  79.             //发送短信消息
  80.             for (int i = 0; i < 10; i++) {
  81.                 String message = "sms inform to user" + i;
  82.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body
  83.                 channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms", null, message.getBytes());
  84.                 System.out.println("Send Message is:'" + message + "'");
  85.             }
  86.  
  87.             //发送邮件和短信消息
  88.             for (int i = 0; i < 10; i++) {
  89.                 String message = "sms and email inform to user" + i;
  90.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body
  91.                 channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms.email", null, message.getBytes());
  92.                 System.out.println("Send Message is:'" + message + "'");
  93.             }
  94.  
  95.         } catch (Exception e) {
  96.             e.printStackTrace();
  97.         }  finally {
  98.             if (channel != null) {
  99.                 try {
  100.                     channel.close();
  101.                 } catch (Exception e) {
  102.                     e.printStackTrace();
  103.                 }
  104.             }
  105.             if (connection != null) {
  106.                 try {
  107.                     connection.close();
  108.                 } catch (IOException e) {
  109.                     e.printStackTrace();
  110.                 }
  111.             }
  112.  
  113.         }
  114.  
  115.  
  116.     }
  117. }
  1. 2）消费者
  1. package com.czxy.xuecheng.rabbitmq.demo04;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  * Created by liangtong.
  9.  */
  10. public class Consumer04_topic_email {
  11.     //队列名称
  12.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
  13.     private static final String EXCHANGE_TOPIC_INFORM = "inform_exchange_topic";
  14.  
  15.     public static void main(String[] args) throws Exception {
  16.         //创建一个与MQ的连接
  17.         ConnectionFactory factory = new ConnectionFactory();
  18.         factory.setHost("127.0.0.1");
  19.         factory.setPort(5672);
  20.         factory.setUsername("guest");
  21.         factory.setPassword("guest");
  22.         //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
  23.         factory.setVirtualHost("/");
  24.  
  25.         //创建一个连接
  26.         Connection connection = factory.newConnection();
  27.         //创建与交换机的通道，每个通道代表一个会话
  28.         Channel channel = connection.createChannel();
  29.         //声明交换机 String exchange, BuiltinExchangeType type
  30.         /**
  31.          * 参数明细
  32.          * 1、交换机名称
  33.          * 2、交换机类型，fanout、topic、direct、headers
  34.          */
  35.         channel.exchangeDeclare(EXCHANGE_TOPIC_INFORM, BuiltinExchangeType.TOPIC);
  36.         //声明队列
  37.         //channel.queueDeclare(String queue, boolean durable, boolean exclusive, booleanautoDelete, Map<String, Object> arguments)
  38.         /**
  39.          * 参数明细：
  40.          * 1、队列名称
  41.          * 2、是否持久化
  42.          * 3、是否独占此队列
  43.          * 4、队列不用是否自动删除
  44.          * 5、参数
  45.          */
  46.         channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
  47.         //交换机和队列绑定String queue, String exchange, String routingKey
  48.         /**
  49.          * 参数明细
  50.          * 1、队列名称
  51.          * 2、交换机名称
  52.          * 3、路由key
  53.          */
  54.         channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPIC_INFORM, QUEUE_INFORM_EMAIL);
  55.         //定义消费方法
  56.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
  57.             @Override
  58.             public void handleDelivery(String consumerTag, Envelope envelope,
  59.                                        AMQP.BasicProperties properties, byte[] body) throws IOException {
  60.                 long deliveryTag = envelope.getDeliveryTag();
  61.                 String exchange = envelope.getExchange();
  62.                 //消息内容
  63.                 String message = new String(body, "UTF-8");
  64.                 System.out.println(message);
  65.             }
  66.         };
  67.         /**
  68.          * 监听队列String queue, boolean autoAck,Consumer callback
  69.          * 参数明细
  70.          * 1、队列名称
  71.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
  72.          为false则需要手动回复
  73.          * 3、消费消息的方法，消费者接收到消息后调用此方法
  74.          */
  75.         channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer);
  76.  
  77.  
  78.     }
  79. }

2.4.5Header

header模式与routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（键值对）匹配队列。

  1. 1）生产者
  1. Map<String, Object> headers_email = new Hashtable<String, Object>();
  2. headers_email.put("inform_type", "email");
  3. Map<String, Object> headers_sms = new Hashtable<String, Object>();
  4. headers_sms.put("inform_type", "sms");
  5. channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
  6. channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_HEADERS_INFORM,"",headers_sms);
  1. 2）发送邮件消费者
  1. channel.exchangeDeclare(EXCHANGE_HEADERS_INFORM, BuiltinExchangeType.HEADERS);
  2. Map<String, Object> headers_email = new Hashtable<String, Object>();
  3. headers_email.put("inform_email", "email");
  4. //交换机和队列绑定
  5. channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
  6. //指定消费队列
  7. channel.basicConsume(QUEUE_INFORM_EMAIL, true, consumer);

2.4.6RPC

RPC即客户端远程调用服务端的方法 ，使用MQ可以实现RPC的异步调用，基于Direct交换机实现，流程如下：

1、客户端即是生产者就是消费者，向RPC请求队列发送RPC调用消息，同时监听RPC响应队列。

2、服务端监听RPC请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果

3、服务端将RPC方法 的结果发送到RPC响应队列

4、客户端（RPC调用方）监听RPC响应队列，接收到RPC调用结果。

2.5SpringBoot整合RabbitMQ

2.5.1搭建环境

  • 创建测试项目：test_rabbitmq_boot

  • 添加依赖
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>xc_test_parent</artifactId>
  7.         <groupId>com.czxy.xuecheng</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  11.  
  12.     <artifactId>test_rabbitmq_boot</artifactId>
  13.  
  14.     <dependencies>
  15.         <dependency>
  16.             <groupId>org.springframework.boot</groupId>
  17.             <artifactId>spring-boot-starter-amqp</artifactId>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.springframework.boot</groupId>
  21.             <artifactId>spring-boot-starter-test</artifactId>
  22.         </dependency>
  23.     </dependencies>
  24.  
  25. </project>
  • 添加yml文件

  1. server:
  2.   port: 8090
  3. spring:
  4.   application:
  5.     name: test_rabbitmq_producer
  6.   rabbitmq:
  7.     host: 127.0.0.1
  8.     port: 5672
  9.     username: guest
  10.     passowrd: guest
  11.     virtualHost: /
  • 创建启动类：TestRabbitMQBootApplication
  1. package com.czxy.xuecheng;
  2.  
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5.  
  6. /**
  7.  * Created by liangtong.
  8.  */
  9. @SpringBootApplication
  10. public class TestRabbitMQBootApplication {
  11.     public static void main(String[] args) {
  12.         SpringApplication.run(TestRabbitMQBootApplication.class, args);
  13.     }
  14. }

2.5.2配置类

  1. package com.czxy.xuecheng.config;
  2.  
  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;
  7.  
  8. /**
  9.  * Created by liangtong.
  10.  */
  11. @Configuration
  12. public class RabbitConfig {
  13.     // 交换机名称
  14.     public static final String EXCHANGE_TOPIC_INFORM = "inform_exchange_topic";
  15.  
  16.     //队列名称
  17.     public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
  18.     public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
  19.  
  20.     /** 交换机配置
  21.      *  ExchangeBuilder提供了fanout、direct、topic、header交换机类型的配置
  22.      *  channel.exchangeDeclare(EXCHANGE_TOPIC_INFORM, BuiltinExchangeType.TOPIC);
  23.      * @return
  24.      */
  25.     @Bean(EXCHANGE_TOPIC_INFORM)
  26.     public Exchange exchange_topic() {
  27.         //durable(true)持久化，消息队列重启后交换机仍然存在
  28.         return ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_INFORM).durable(true).build();
  29.     }
  30.  
  31.     /**
  32.      * 声明队列
  33.      * channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
  34.      * @return
  35.      */
  36.     @Bean(QUEUE_INFORM_SMS)
  37.     public Queue queue_inform_sms(){
  38.         return new Queue(QUEUE_INFORM_SMS);
  39.     }
  40.     @Bean(QUEUE_INFORM_EMAIL)
  41.     public Queue queue_inform_email(){
  42.         return new Queue(QUEUE_INFORM_EMAIL,true,false,false);
  43.     }
  44.  
  45.     /**
  46.      * 绑定队列到交换机
  47.      * channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPIC_INFORM, "inform.#.email.#");
  48.      * @param queue
  49.      * @param exchange
  50.      * @return
  51.      */
  52.     @Bean
  53.     public Binding binding_queue_inform_sms(@Qualifier(QUEUE_INFORM_SMS) Queue queue, @Qualifier(EXCHANGE_TOPIC_INFORM) Exchange exchange) {
  54.         return BindingBuilder.bind(queue).to(exchange).with("inform.#.sms.#").noargs();
  55.     }
  56.     @Bean
  57.     public Binding binding_queue_inform_email(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue, @Qualifier(EXCHANGE_TOPIC_INFORM) Exchange exchange) {
  58.         return BindingBuilder.bind(queue).to(exchange).with("inform.#.email.#").noargs();
  59.     }
  60.  
  61. }

2.5.3生产者

  1. package com.czxy.xuecheng;
  2.  
  3. import com.czxy.xuecheng.config.RabbitConfig;
  4. import org.junit.Test;
  5. import org.junit.runner.RunWith;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.boot.test.context.SpringBootTest;
  8. import org.springframework.test.context.junit4.SpringRunner;
  9.  
  10. import javax.annotation.Resource;
  11.  
  12. /**
  13.  * Created by liangtong.
  14.  */
  15. @RunWith(SpringRunner.class)
  16. @SpringBootTest(classes = TestRabbitMQBootApplication.class)
  17. public class Producer05Topic {
  18.  
  19.     @Resource
  20.     private RabbitTemplate rabbitTemplate;
  21.  
  22.     @Test
  23.     public void testSendEmail() {
  24.         //channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.email", null, message.getBytes());
  25.         for(int i = 0 ; i < 5 ; i ++) {
  26.             String message = "email inform to user" + i;
  27.             rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_TOPIC_INFORM,"inform.email",message);
  28.             System.out.println("Send Message is:'" + message + "'");
  29.         }
  30.     }
  31.  
  32.     @Test
  33.     public void testSendSms() {
  34.         //channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms", null, message.getBytes());
  35.         for(int i = 0 ; i < 5 ; i ++) {
  36.             String message = "sms inform to user" + i;
  37.             rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_TOPIC_INFORM,"inform.sms",message);
  38.             System.out.println("Send Message is:'" + message + "'");
  39.         }
  40.     }
  41.  
  42.     @Test
  43.     public void testSendSmsAndEmail() {
  44.         //channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms.email", null, message.getBytes());
  45.         for(int i = 0 ; i < 5 ; i ++) {
  46.             String message = "sms and email inform to user" + i;
  47.             rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_TOPIC_INFORM,"inform.sms.email",message);
  48.             System.out.println("Send Message is:'" + message + "'");
  49.         }
  50.     }
  51.  
  52. }

2.5.4消费者

  1. package com.czxy.xuecheng.listener;
  2.  
  3. import com.czxy.xuecheng.config.RabbitConfig;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  6. import org.springframework.stereotype.Component;
  7.  
  8. /**
  9.  * Created by liangtong.
  10.  */
  11. @Component
  12. public class Consumer05Topic {
  13.  
  14.     @RabbitListener(queues = RabbitConfig.QUEUE_INFORM_EMAIL)
  15.     public void receiveEmail(String msg , Message message){
  16.         System.out.println("receive message is:" + msg);
  17.     }
  18. /*
  19.     @RabbitListener(queues = RabbitConfig.QUEUE_INFORM_SMS)
  20.     public void receiveSmS(String msg , Message message){
  21.         System.out.println("receive message is:" + msg);
  22.     }
  23.     */
  24. }

网站建设定制开发 软件系统开发定制 定制软件开发 软件开发定制 定制app开发 app开发定制 app开发定制公司 电商商城定制开发 定制小程序开发 定制开发小程序 客户管理系统开发定制 定制网站 定制开发 crm开发定制 开发公司 小程序开发定制 定制软件 收款定制开发 企业网站定制开发 定制化开发 android系统定制开发 定制小程序开发费用 定制设计 专注app软件定制开发 软件开发定制定制 知名网站建设定制 软件定制开发供应商 应用系统定制开发 软件系统定制开发 企业管理系统定制开发 系统定制开发