devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ 交换机类型

作者：我家自动化 | 2024-05-20 11:26:23

赞

踩

RabbitMQ 交换机类型

常用交换机

发布订阅（Publish/Subscribe）交换机

 一个生产者给多个队列发送消息，X 代表交换机。

交换机的作用：类似网络路由器，主要提供转发功能，解决怎么把消息转发到不同的队列中，让消费者从不同队列取然后消费。

绑定：交换机和队列关联起来

发布订阅交换机，队列进行持久化，生产者发布消息，所有消费者都能接收到消息。

生产者代码

  1. public class FanoutProducer{
  2.  
  3.   private static final String EXCHANGE_NAME = "fanout_exchange";
  4.  
  5.   public static void main(String[] argv) throws Exception {
  6.     ConnectionFactory factory = new ConnectionFactory();
  7.     factory.setHost("localhost");
  8.     try (Connection connection = factory.newConnection();
  9.          Channel channel = connection.createChannel()) {
  10.         //创建交换机，参数：交换机名称，交换机类型
  11.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
  12.         Scanner scanner = new Scanner(System.in);
  13.         while(scanner.hasNext()){
  14.             String message = scanner.nextLine();
  15.             //第二个参数是路由规则
  16.             channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
  17.             System.out.println(" [x] Sent '" + message + "'");
  18.         }
  19.  
  20.     }
  21.   }
  22. }

消费者代码

  1. public class FanoutConsumer {
  2.     private static final String EXCHANGE_NAME = "fanout_exchange";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         factory.setHost("localhost");
  7.         Connection connection = factory.newConnection();
  8.         Channel channel = connection.createChannel();
  9.         //绑定交换机，以及设置绑定规则
  10.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
  11.         String queueName1 = "xiaowang";
  12.         String queueName2 = "xiaoli";
  13.         channel.queueDeclare(queueName1, true, false, false, null);
  14.         channel.queueDeclare(queueName2, true, false, false, null);
  15.         //创建队列，不指定队列，随机分配
  16.         //String queueName = channel.queueDeclare().getQueue();
  17.         channel.queueBind(queueName1, EXCHANGE_NAME, "");
  18.         System.out.println(" [xiaowang] Waiting for messages. To exit press CTRL+C");
  19. 		//交换机绑定队列
  20.         channel.queueBind(queueName2, EXCHANGE_NAME, "");
  21.         System.out.println(" [xiaoli] Waiting for messages. To exit press CTRL+C");
  22.         DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
  23.             String message = new String(delivery.getBody(), "UTF-8");
  24.             System.out.println(  " [xiaowang] Received '" + message + "'");
  25.         };
  26.         DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
  27.             String message = new String(delivery.getBody(), "UTF-8");
  28.             System.out.println(  " [xiaoli] Received '" + message + "'");
  29.         };
  30.         channel.basicConsume(queueName1, true, deliverCallback1, consumerTag -> {
  31.         });
  32.         channel.basicConsume(queueName2, true, deliverCallback2, consumerTag -> {
  33.         });
  34.     }
  35. }

channel 频道：理解为操作消息队列的 Client，通过 channel 收发消息，提供了和消息对了 server 建立通信的传输方法

channel.queueDeclare 方法参数：

queue：这是一个字符串参数，代表要声明的队列的名称。如果队列不存在，则会自动创建一个新的队列。

durable：这是一个布尔值参数，表示队列是否持久化。如果设置为true，则队列会在服务器重启后仍然存在；如果设置为false，则队列在服务器重启后会被删除。默认值为false。

exclusive：这也是一个布尔值参数，表示队列是否为独占模式。如果设置为true，则只有当前连接可以访问该队列；如果设置为false，则其他连接也可以访问该队列。默认值为false。

autoDelete：这是另一个布尔值参数，表示队列是否自动删除。如果设置为true，则当最后一个消费者取消订阅时，队列将被删除；如果设置为false，则队列将一直存在，直到手动删除或服务器重启。默认值为false。

arguments：这是一个可选参数，用于设置队列的其他属性，比如消息的最大长度、最大优先级等。

channel.basicPublish 参数：

exchange：这是一个字符串参数，代表交换机的名称。如果不需要使用特定的交换机，可以传递一个空字符串（""）。交换机是RabbitMQ中用于接收生产者发送的消息并根据绑定规则路由到队列的组件。

routingKey：这也是一个字符串参数，它指定了发布消息的队列。无论通道绑定到哪个队列，最终发布的消息都会包含这个指定的路由键。路由键是用来确定消息应该发送到哪个队列的重要信息。

message：这是要发布的消息本身，通常是字节数组的形式。

properties：这是一个可选参数，用于设置消息的属性，比如消息的优先级、过期时间等。

在使用channel.basicPublish时，需要注意以下几点：

exchange和routingKey不能为空：在AMQImpl类中的实现要求这两个参数都不能为null，否则会抛出异常。

交换机类型：根据不同的需求，可以选择不同类型的交换机，如fanout、direct或topic。每种类型的交换机都有其特定的路由规则。

非命名队列：在某些情况下，比如日志系统，可以使用非命名队列，这样消费者可以接收到所有相关的日志消息，而不是特定的部分。

 channel.basicConsume 参数：

queue：这是一个字符串参数，代表要消费的队列的名称。如果队列不存在，则会抛出异常。
onMessage：这是一个回调函数，当有新的消息到达时会被调用。该函数需要接收两个参数：一个表示消息内容的Delivery对象和一个表示通道的Channel对象。
consumerTag：这是一个可选参数，用于标识消费者。如果没有指定，则会自动生成一个唯一的标识符。
autoAck：这是一个布尔值参数，表示是否自动确认消息。如果设置为true，则在消息被处理后会自动发送确认信息；如果设置为false，则需要手动发送确认信息。默认值为false。
arguments：这是一个可选参数，用于设置消费者的其他属性，比如消息的最大长度、最大优先级等。
在使用channel.basicConsume时，需要注意以下几点：

队列名称：队列名称应该是唯一的，否则会抛出异常。
消息处理：在onMessage回调函数中，需要对消息进行处理，并根据需要发送确认信息。
消费者标识符：可以通过设置consumerTag来标识消费者，以便在后续操作中进行识别和管理。
消费者属性：可以通过设置消费者的其他属性来控制消费者的行为，比如设置消息的最大长度、最大优先级等。

路由交换机 （Direct exchange)

和订阅发布的区别是在交换机和队列之间有一个路由键，用来控制消息发送到哪个队列中供消费者消费。生产者给交换机一个标识，让交换机给指定的队列转发消息。

生产者代码

  1. public class DirectProducer {
  2.  
  3.     private static final String EXCHANGE_NAME = "direct_exchange";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         ConnectionFactory factory = new ConnectionFactory();
  7.         factory.setHost("localhost");
  8.         try (Connection connection = factory.newConnection();
  9.              Channel channel = connection.createChannel()) {
  10.             //创建交换机，交换机类型是 direct
  11.             channel.exchangeDeclare(EXCHANGE_NAME, "direct");
  12.             Scanner scanner = new Scanner(System.in);
  13.             while (scanner.hasNext()){
  14.                 String userInput = scanner.nextLine();
  15.                 //输入的时候带着标识，标识就是路由键
  16.                 String[] strs = userInput.split(" ");
  17.                 if(strs.length<1){
  18.                     continue;
  19.                 }
  20.                 //消息
  21.                 String message = strs[0];
  22.                 //路由键
  23.                 String severity = strs[1];
  24.                 //发送消息时带着路由键
  25.                 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));
  26.                 System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
  27.             }
  28.         }
  29.     }
  30. }

消费者代码

  1. public class DirectConsumer {
  2.  
  3.     private static final String EXCHANGE_NAME = "direct_exchange";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         ConnectionFactory factory = new ConnectionFactory();
  7.         factory.setHost("localhost");
  8.         Connection connection = factory.newConnection();
  9.         Channel channel = connection.createChannel();
  10.         channel.exchangeDeclare(EXCHANGE_NAME, "direct");
  11.         String queueName1 = "xiaohong";
  12.         String queueName2 = "xiaobai";
  13.         channel.queueDeclare(queueName1, true, false, false, null);
  14.         channel.queueDeclare(queueName2, true, false, false, null);
  15.         //交换机使用路由键绑定队列
  16.         channel.queueBind(queueName1, EXCHANGE_NAME, "xiaohong");
  17.         channel.queueBind(queueName2, EXCHANGE_NAME, "xiaobai");
  18.         System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
  19.  
  20.         DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
  21.             String message = new String(delivery.getBody(), "UTF-8");
  22.             System.out.println(" [xiaohong] Received '" +
  23.                     delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  24.         };
  25.         DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
  26.             String message = new String(delivery.getBody(), "UTF-8");
  27.             System.out.println(" [xiaobai] Received '" +
  28.                     delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  29.         };
  30.         channel.basicConsume(queueName1, true, deliverCallback1, consumerTag -> {
  31.         });
  32.         channel.basicConsume(queueName2, true, deliverCallback2, consumerTag -> {
  33.         });
  34.     }
  35. }

主题交换机 （Topic exchange）

在路由交换机的基础上，消息会具有一个模糊的路由键转发给指定的对俄（一系列的路由键、一类的路由键）

1. （*）标识匹配一个单词，比如 *.orange 表示 a.orange b.orange 都能匹配

2. （#）表示 0 个或多个单词，比如 a,#, a.a, a.b 都可以

生产者代码

  1. public class TopicProduce {
  2.  
  3.   private static final String EXCHANGE_NAME = "topic_exchange1";
  4.  
  5.   public static void main(String[] argv) throws Exception {
  6.       ConnectionFactory factory = new ConnectionFactory();
  7.       factory.setHost("localhost");
  8.       try (Connection connection = factory.newConnection();
  9.            Channel channel = connection.createChannel()) {
  10.           channel.exchangeDeclare(EXCHANGE_NAME, "topic");
  11.  
  12.           Scanner scanner = new Scanner(System.in);
  13.           while (scanner.hasNext()) {
  14.               String userInput = scanner.nextLine();
  15.               String[] strs = userInput.split(" ");
  16.               if (strs.length < 1) {
  17.                   continue;
  18.               }
  19.               //消息
  20.               String message = strs[0];
  21.               //路由键
  22.               String severity = strs[1];
  23.  
  24.               channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));
  25.               System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
  26.           }
  27.       }
  28.   }
  29. }

消费者代码

  1. public class TopicConsumer {
  2.  
  3.     private static final String EXCHANGE_NAME = "topic_exchange1";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         ConnectionFactory factory = new ConnectionFactory();
  7.         factory.setHost("localhost");
  8.         Connection connection = factory.newConnection();
  9.         Channel channel = connection.createChannel();
  10.  
  11.         channel.exchangeDeclare(EXCHANGE_NAME, "topic");
  12.         channel.exchangeDeclare(EXCHANGE_NAME, "topic");
  13.         channel.exchangeDeclare(EXCHANGE_NAME, "topic");
  14.         String queueName1 = "xiaohei";
  15.         String queueName2 = "xiaolv";
  16.         String queueName3 = "xiaohuang";
  17.         channel.queueDeclare(queueName1, true, false, false, null);
  18.         channel.queueDeclare(queueName2, true, false, false, null);
  19.         channel.queueDeclare(queueName3, true, false, false, null);
  20.         //交换机使用路由键绑定队列，路由键绑定在第三个参数
  21.         channel.queueBind(queueName1, EXCHANGE_NAME, "#.前端.#");
  22.         channel.queueBind(queueName2, EXCHANGE_NAME, "#.后端.#");
  23.         channel.queueBind(queueName2, EXCHANGE_NAME, "#.产品.#");
  24.         System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
  25.         //收到消息后如何处理
  26.         DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
  27.             String message = new String(delivery.getBody(), "UTF-8");
  28.             System.out.println(" [xiaohei] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  29.         };
  30.         DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
  31.             String message = new String(delivery.getBody(), "UTF-8");
  32.             System.out.println(" [xiaolv] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  33.         };
  34.         DeliverCallback deliverCallback3 = (consumerTag, delivery) -> {
  35.             String message = new String(delivery.getBody(), "UTF-8");
  36.             System.out.println(" [xiaohuang] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  37.         };
  38.         channel.basicConsume(queueName1, true, deliverCallback1, consumerTag -> {
  39.         });
  40.         channel.basicConsume(queueName2, true, deliverCallback2, consumerTag -> {
  41.         });
  42.         channel.basicConsume(queueName3, true, deliverCallback3, consumerTag -> {
  43.         });
  44.  
  45.  
  46.     }
  47. }

核心机制

消息过期机制

官方文档：Preventing Unbounded Buffers with RabbitMQ | RabbitMQ

每个消息指定一个有效期，一段时间内没有被消费者处理，就过期了。

比如消费者挂了，消息一直不被处理，订单就会失效。

可以清理过期的数据，模拟延迟队列的实现。

给每条消息都设置过期时间：

  1. Map<String, Object> args = new HashMap<String, Object>();
  2. args.put("x-message-ttl", 60000);
  3. channel.queueDeclare(QUEUE_NAME, false, false, false, args);

给队列设置过期时间，设置在生产者中

  1. AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
  2.         .expiration("1000")
  3.         .build();

生产者代码

  1. public class TtlProducer {
  2.  
  3.   private final static String QUEUE_NAME = "Tll_queue";
  4.   public static void main(String[] argv) throws Exception {
  5.     //创建连接
  6.     ConnectionFactory factory = new ConnectionFactory();
  7.     //设置了本地连接，如果修改了用户名和密码，需要设置
  8.     /*factory.setPassword();
  9.     factory.setUsername();*/
  10.     factory.setHost("localhost");
  11.     //建立连接、创建频道
  12.     //频道，类似客户端，用于调用server
  13.     Connection connection = factory.newConnection();
  14.     Channel channel = connection.createChannel();
  15.     String message = "Hello World!";
  16.     //发消息设置过期时间
  17.     AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
  18.             .expiration("1000")
  19.             .build();
  20.     //发送消息
  21.     channel.basicPublish("",QUEUE_NAME,properties,message.getBytes(StandardCharsets.UTF_8));
  22.     System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
  23.  
  24.   }
  25. }

消费者代码

  1. public class TtlConsumer {
  2.  
  3.   private final static String QUEUE_NAME = "Tll_queue";
  4.  
  5.   public static void main(String[] argv) throws Exception {
  6.     ConnectionFactory factory = new ConnectionFactory();
  7.     factory.setHost("localhost");
  8.     Connection connection = factory.newConnection();
  9.     Channel channel = connection.createChannel();
  10.     //声明队列，同一个消息队列参数必须一致
  11.     Map<String, Object> args = new HashMap<String, Object>();
  12.     args.put("x-message-ttl", 60000);
  13.     channel.queueDeclare(QUEUE_NAME, false, false, false, args);
  14.     //定义了如何处理消息
  15.     DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  16.       String message = new String(delivery.getBody(), "UTF-8");
  17.       System.out.println(" [x] Received '" + message + "'");
  18.     };
  19.     //接收、消费消息 第二个参数 autoAck
  20.     channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
  21.     System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
  22.  
  23.   }
  24. }

死信队列

官方文档：Dead Letter Exchanges | RabbitMQ

为了保证消息的可靠性，比如每条消息都成功消费，需要提供一个容错机制，即：失效的消息怎么办？

死信：过期的消息、拒收的消息、消息队列满了、处理失败的消息的统称。

死信队列：处理死信的队列。

死信交换机：专门给死信队列转发消息的，存在路由绑定关系

实际就是设置一个普通的队列，专门将死信发送到这个队列中处理。

1. 创建死信交换机和死信队列

  1. //声明死信交换机
  2. String queueName = "laoban_dlx_queue";
  3. channel.queueDeclare(queueName, true, false, false, null);
  4. channel.queueBind(queueName, EXCHANGE_NAME, "laoban");

2. 给失败之后需要容错处理的队列绑定死信交换机

args2.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);

3. 绑定交换机到死信队列

args2.put("x-dead-letter-routing-key", "waibao");

生产者代码

  1. public class DLXDirectProducer {
  2.  
  3.     private static final String EXCHANGE_NAME = "direct2_exchange";
  4.     private static final String DEAD_EXCHANGE_NAME = "dlx_direct2_exchange";
  5.  
  6.     public static void main(String[] argv) throws Exception {
  7.         ConnectionFactory factory = new ConnectionFactory();
  8.         factory.setHost("localhost");
  9.         try (Connection connection = factory.newConnection();
  10.              Channel channel = connection.createChannel()) {
  11.             //声明死信交换机
  12.             channel.exchangeDeclare(DEAD_EXCHANGE_NAME, "direct");
  13.             String queueName = "laoban_dlx_queue";
  14.             channel.queueDeclare(queueName, true, false, false, null);
  15.             channel.queueBind(queueName, DEAD_EXCHANGE_NAME, "laoban");
  16.  
  17.             String queueName2 = "waibao_dlx_queue";
  18.             channel.queueDeclare(queueName2, true, false, false, null);
  19.             channel.queueBind(queueName2, DEAD_EXCHANGE_NAME, "waibao");
  20.             channel.exchangeDeclare(EXCHANGE_NAME, "direct");
  21.             DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
  22.                 String message = new String(delivery.getBody(), "UTF-8");
  23.                 System.out.println(" [laoban] Received '" +
  24.                         delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  25.             };
  26.             DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
  27.                 String message = new String(delivery.getBody(), "UTF-8");
  28.                 System.out.println(" [waibao] Received '" +
  29.                         delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  30.             };
  31.             channel.basicConsume(queueName, false, deliverCallback1, consumerTag -> {
  32.             });
  33.             channel.basicConsume(queueName2, false, deliverCallback2, consumerTag -> {
  34.             });
  35.             Scanner scanner = new Scanner(System.in);
  36.             while (scanner.hasNext()){
  37.                 String userInput = scanner.nextLine();
  38.                 String[] strs = userInput.split(" ");
  39.                 if(strs.length<1){
  40.                     continue;
  41.                 }
  42.                 //消息
  43.                 String message = strs[0];
  44.                 //路由键
  45.                 String severity = strs[1];
  46.  
  47.                 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));
  48.                 System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
  49.             }
  50.  
  51.         }
  52.     }
  53. }

消费者代码

  1. public class DLXDirectConsumer {
  2.  
  3.     private static final String EXCHANGE_NAME = "direct2_exchange";
  4.     private static final String DEAD_EXCHANGE_NAME = "dlx_direct2_exchange";
  5.  
  6.  
  7.     public static void main(String[] argv) throws Exception {
  8.         ConnectionFactory factory = new ConnectionFactory();
  9.         factory.setHost("localhost");
  10.         Connection connection = factory.newConnection();
  11.         Channel channel = connection.createChannel();
  12.         channel.exchangeDeclare(EXCHANGE_NAME, "direct");
  13.         //死信交换机绑定工作队列，当信息错误就从工作队列发送到死信交换机
  14.         Map<String, Object> args1 = new HashMap<String, Object>();
  15.         //指定绑定哪个交换机
  16.         args1.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
  17.         //死信要发送到哪个队列
  18.         args1.put("x-dead-letter-routing-key", "laoban");
  19.         String queueName1 = "doghuang";
  20.         channel.queueDeclare(queueName1, true, false, false, args1);
  21.         channel.queueBind(queueName1, EXCHANGE_NAME, "doghuang");
  22.  
  23.         //绑定cat 队列
  24.  
  25.         String queueName2 = "catbai";
  26.         Map<String, Object> args2 = new HashMap<String, Object>();
  27.         args2.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
  28.         args2.put("x-dead-letter-routing-key", "waibao");
  29.         channel.queueDeclare(queueName2, true, false, false, args2);
  30.         channel.queueBind(queueName2, EXCHANGE_NAME, "catbai");
  31.  
  32.  
  33.         //交换机使用路由键绑定队列
  34.         System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
  35.  
  36.         DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
  37.             String message = new String(delivery.getBody(), "UTF-8");
  38.             channel.basicNack(delivery.getEnvelope().getDeliveryTag(),false,false);
  39.             System.out.println(" [doghuang] Received '" +
  40.                     delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  41.         };
  42.         DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
  43.             String message = new String(delivery.getBody(), "UTF-8");
  44.             System.out.println(" [catbai] Received '" +
  45.                     delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
  46.         };
  47.         channel.basicConsume(queueName1, false, deliverCallback1, consumerTag -> {
  48.         });
  49.         channel.basicConsume(queueName2, false, deliverCallback2, consumerTag -> {
  50.         });
  51.  
  52.     }
  53. }

项目实战

项目中使用可以选择两种方法

1. 官方的客户端，兼容性好，灵活，需要自己维护管理

2. 使用封装好的客户端，比如 Spring Boot RabbitMQ Starter

优点：简单易用

缺点：不够灵活，被框架限制

小项目使用封装好的足够

1. 依赖引入

引入和自己 Spring Boot 版本相同的依赖，避免出现不能运行的错误

  1. <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
  2. <dependency>
  3.     <groupId>org.springframework.boot</groupId>
  4.     <artifactId>spring-boot-starter-amqp</artifactId>
  5.     <version>2.6.13</version>
  6. </dependency>

2. 引入配置

  1. rabbitmq:
  2.   host: localhost
  3.   port: 5672
  4.   password: guest
  5.   username: guset

3. 创建交换机和消息队列，这个只需要启动一次创建即可

  1. /**
  2.  * 只启动一次，创建交换机和消息队列
  3.  */
  4. public class MqInitMain {
  5.  
  6.     private static final String EXCHANGE_NAME = "code_exchange";
  7.     public static void main(String[] args) throws Exception {
  8.         ConnectionFactory factory = new ConnectionFactory();
  9.         factory.setHost("localhost");
  10.         Connection connection = factory.newConnection();
  11.         Channel channel = connection.createChannel();
  12.         channel.exchangeDeclare(EXCHANGE_NAME, "direct");
  13.  
  14.         //绑定一个队列
  15.         String queueName = "code_queue";
  16.         channel.queueDeclare(queueName,true,false,false,null);
  17.         channel.queueBind(queueName,EXCHANGE_NAME,"BI_routingKey");
  18.     }
  19. }

4. 生产者

  1. /**
  2.  * 生产者
  3.  */
  4. @Component
  5. public class MyMessageProducer {
  6.  
  7.  
  8.     @Resource
  9.     private RabbitTemplate rabbitTemplate;
  10.  
  11.     //1.交换机名称2. 交换机路由键3.发送的消息
  12.     public void sendMessage(String exchange, String routingKey,String message){
  13.         rabbitTemplate.convertAndSend(exchange,routingKey,message);
  14.     }
  15.  
  16. }

5. 消费者

  1. /**
  2.  * 消费者
  3.  */
  4. @Component
  5. @Slf4j
  6. public class MessageConsumer {
  7.  
  8.     @RabbitListener(queues = {"code_queue"},ackMode = "MANUAL")
  9.     public void receiveMessage(String message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long delivery){
  10.         log.info("receiveMessage message={}",message);
  11.         try {
  12.             channel.basicAck(delivery,false);
  13.         } catch (IOException e) {
  14.             throw new RuntimeException(e);
  15.         }
  16.     }
  17. }

6. 测试

  1. @SpringBootTest
  2. class MyMessageProducerTest {
  3.  
  4.     @Resource
  5.     private MyMessageProducer myMessageProducer;
  6.     @Test
  7.     void sendMessage() {
  8.         myMessageProducer.sendMessage("code_exchange","BI_routingKey","你好吗");
  9.     }
  10. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/597462
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号