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

java 使用RabbitMQ_java使用rabbitmq

作者：花生_TL007 | 2024-04-08 09:16:40

赞

踩

java使用rabbitmq

简单模式：生产者发布Publish消息到队列，消费者从队列消费Consume消息

生产者代码：

  1. package com.example.simple;
  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. public class SimpleProvider {
  11.     public static void main(String[] args) throws IOException, TimeoutException {
  12.         //创建链接工厂对象
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         //设置RabbitMQ服务主机地址，默认localhost
  15.         connectionFactory.setHost("localhost");
  16.         //设置端口，默认5672
  17.         connectionFactory.setPort(5672);
  18.         //设置虚拟主机名，默认/
  19.         connectionFactory.setVirtualHost("/mytest");
  20.         //设置用户名密码
  21.         connectionFactory.setUsername("test");
  22.         connectionFactory.setPassword("123456");
  23.  
  24.         //创建链接
  25.         Connection connection = connectionFactory.newConnection();
  26.         //创建通道
  27.         Channel channel = connection.createChannel();
  28.         //声明队列
  29.         //参数1：消息队列名称
  30.         //参数2，是否持久化
  31.         //参数3，是否独占本次connection链接
  32.         //参数4，是否自动删除消息
  33.         //参数5，是否需要传递额外的队列数据Map
  34.         channel.queueDeclare("simple_queue1",true,false,false,null);
  35.         //创建消息
  36.         String msg = "hello";
  37.         //发布消息
  38.         //参数1，使用指定的交换机，不设置则使用默认交换机
  39.         //参数2，指定路由器key，如果是简单模式直接给队列名就行
  40.         //参数3，发生消息时是否需要额外的消息数据
  41.         //参数4，消息内容
  42.         channel.basicPublish("","simple_queue1",null,msg.getBytes());
  43.         //关闭资源
  44.         channel.close();
  45.         connection.close();
  46.     }
  47. }

 消费者代码

  1. package com.example.simple;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. public class SimpleConsumer {
  9.     public static void main(String[] args) throws IOException, TimeoutException {
  10.         //创建链接工厂对象
  11.         ConnectionFactory connectionFactory = new ConnectionFactory();
  12.         //设置RabbitMQ服务主机地址，默认localhost
  13.         connectionFactory.setHost("localhost");
  14.         //设置端口，默认5672
  15.         connectionFactory.setPort(5672);
  16.         //设置虚拟主机名，默认/
  17.         connectionFactory.setVirtualHost("/mytest");
  18.         //设置用户名密码
  19.         connectionFactory.setUsername("test");
  20.         connectionFactory.setPassword("123456");
  21.  
  22.         //创建链接
  23.         Connection connection = connectionFactory.newConnection();
  24.         //创建频道
  25.         Channel channel = connection.createChannel();
  26.  
  27.         //声明队列
  28.         //参数1：消息队列名称
  29.         //参数2，是否持久化
  30.         //参数3，是否独占本次connection链接
  31.         //参数4，是否自动删除消息
  32.         //参数5，是否需要传递额外的队列数据Map
  33.         channel.queueDeclare("simple_queue1",true,false,false,null);
  34.         //创建消费者，并且设置消息处理
  35.         //指定频道
  36.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
  37.             /**
  38.              *
  39.              * @param consumerTag 消费者名称，如果之前没有指定，则会自动生成一个
  40.              * @param envelope  消息一些额外的参数 channel.basicPublish("","simple_queue1",null,msg.getBytes());
  41.              * @param properties  另外的数据所封装的属性对象 null
  42.              * @param body  消息本身
  43.              * @throws IOException
  44.              */
  45.             @Override
  46.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  47.                 //实现消费的业务逻辑
  48.                 System.out.println("consumerTag"+consumerTag);
  49.                 System.out.println("交换机："+envelope.getExchange()+"----RoutingKey："+envelope.getRoutingKey()+"----DeliveryTag："+envelope.getDeliveryTag());
  50.                 System.out.println("消息本身："+new String(body,"UTF-8"));
  51.             }
  52.         };
  53.         //监听消息
  54.         //参数1，指定要监听的队列的名称
  55.         //参数2，设置消息的应答模式，true自动应答（性能好，消息有可能丢失），false手动应答
  56.         //参数3，
  57.         channel.basicConsume("simple_queue1",true,defaultConsumer);
  58.  
  59.  
  60.         //关闭资源，建议不关闭一直监听
  61.     }
  62. }

 工作模式：多个消费者消费多个不同的消息，创建多个消费者即可

订阅-广播模式：同一个消息被多个消费者消费。上面两个模式只有三个角色分别是生产者，队列，消费者。而在订阅模式中多了一个角色，那就是exchange交换机角色。

订阅模式大概过程是：生产者生产消息后，发送给交换机，然后由交换机转发给各个队列，在这之前队列需要和交换机进行绑定。     .

 生产者代码

  1. package com.example.fanout;
  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. import java.util.concurrent.TimeoutException;
  10.  
  11. public class FanoutProvider {
  12.     public static void main(String[] args) throws IOException, TimeoutException {
  13.         //创建链接工厂对象
  14.         ConnectionFactory connectionFactory = new ConnectionFactory();
  15.         //设置RabbitMQ服务主机地址，默认localhost
  16.         connectionFactory.setHost("localhost");
  17.         //设置端口，默认5672
  18.         connectionFactory.setPort(5672);
  19.         //设置虚拟主机名，默认/
  20.         connectionFactory.setVirtualHost("/mytest");
  21.         //设置用户名密码
  22.         connectionFactory.setUsername("test");
  23.         connectionFactory.setPassword("123456");
  24.  
  25.         //创建链接
  26.         Connection connection = connectionFactory.newConnection();
  27.         //创建通道
  28.         Channel channel = connection.createChannel();
  29.         //声明队列
  30.         //参数1：消息队列名称
  31.         //参数2，是否持久化
  32.         //参数3，是否独占本次connection链接
  33.         //参数4，是否自动删除消息
  34.         //参数5，是否需要传递额外的队列数据Map
  35.         channel.queueDeclare("fanout_queue1",true,false,false,null);
  36.         //声明两个队列
  37.         channel.queueDeclare("fanout_queue2",true,false,false,null);
  38.  
  39.         //声明交换机
  40.         //参数1 交换机名称
  41.         //参数2，指定交换机的类型 BuiltinExchangeType.FANOUT（广播）
  42.         channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);
  43.         //将队列绑定到指定的交换机
  44.         //参数1，指定要绑定的队列
  45.         //参数2，指定绑定的交换机
  46.         //参数3，指定routingkey，广播模式可以不写
  47.         channel.queueBind("fanout_queue1","fanout_exchange","");
  48.         channel.queueBind("fanout_queue2","fanout_exchange","");
  49.  
  50.         //创建消息
  51.         String msg = "hello";
  52.         //发布消息
  53.         //参数1，使用指定的交换机，不设置则使用默认交换机
  54.         //参数2，指定路由器key，如果是简单模式直接给队列名就行，广播模式指定为空
  55.         //参数3，发生消息时是否需要额外的消息数据
  56.         //参数4，消息内容
  57.         channel.basicPublish("fanout_exchange","",null,msg.getBytes());
  58.         //关闭资源
  59.         channel.close();
  60.         connection.close();
  61.     }
  62. }

消费者1

  1. package com.example.fanout;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. public class FanoutConsumer {
  9.     public static void main(String[] args) throws IOException, TimeoutException {
  10.         //创建链接工厂对象
  11.         ConnectionFactory connectionFactory = new ConnectionFactory();
  12.         //设置RabbitMQ服务主机地址，默认localhost
  13.         connectionFactory.setHost("localhost");
  14.         //设置端口，默认5672
  15.         connectionFactory.setPort(5672);
  16.         //设置虚拟主机名，默认/
  17.         connectionFactory.setVirtualHost("/mytest");
  18.         //设置用户名密码
  19.         connectionFactory.setUsername("test");
  20.         connectionFactory.setPassword("123456");
  21.  
  22.         //创建链接
  23.         Connection connection = connectionFactory.newConnection();
  24.         //创建频道
  25.         Channel channel = connection.createChannel();
  26.  
  27.         //声明队列
  28.         //参数1：消息队列名称
  29.         //参数2，是否持久化
  30.         //参数3，是否独占本次connection链接
  31.         //参数4，是否自动删除消息
  32.         //参数5，是否需要传递额外的队列数据Map
  33.         channel.queueDeclare("fanout_queue1",true,false,false,null);
  34.         //创建消费者，并且设置消息处理
  35.         //指定频道
  36.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
  37.             /**
  38.              *
  39.              * @param consumerTag 消费者名称，如果之前没有指定，则会自动生成一个
  40.              * @param envelope  消息一些额外的参数 channel.basicPublish("","simple_queue1",null,msg.getBytes());
  41.              * @param properties  另外的数据所封装的属性对象 null
  42.              * @param body  消息本身
  43.              * @throws IOException
  44.              */
  45.             @Override
  46.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  47.                 //实现消费的业务逻辑
  48.                 System.out.println("consumerTag"+consumerTag);
  49.                 System.out.println("交换机："+envelope.getExchange()+"----RoutingKey："+envelope.getRoutingKey()+"----DeliveryTag："+envelope.getDeliveryTag());
  50.                 System.out.println("消息本身："+new String(body,"UTF-8"));
  51.             }
  52.         };
  53.         //监听消息
  54.         //参数1，指定要监听的队列的名称
  55.         //参数2，设置消息的应答模式，true自动应答（性能好，消息有可能丢失），false手动应答
  56.         //参数3，
  57.         channel.basicConsume("fanout_queue1",true,defaultConsumer);
  58.  
  59.  
  60.         //关闭资源，建议不关闭一直监听
  61.     }
  62. }

消费者2

  1. package com.example.fanout;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. public class FanoutConsumer2 {
  9.     public static void main(String[] args) throws IOException, TimeoutException {
  10.         //创建链接工厂对象
  11.         ConnectionFactory connectionFactory = new ConnectionFactory();
  12.         //设置RabbitMQ服务主机地址，默认localhost
  13.         connectionFactory.setHost("localhost");
  14.         //设置端口，默认5672
  15.         connectionFactory.setPort(5672);
  16.         //设置虚拟主机名，默认/
  17.         connectionFactory.setVirtualHost("/mytest");
  18.         //设置用户名密码
  19.         connectionFactory.setUsername("test");
  20.         connectionFactory.setPassword("123456");
  21.  
  22.         //创建链接
  23.         Connection connection = connectionFactory.newConnection();
  24.         //创建频道
  25.         Channel channel = connection.createChannel();
  26.  
  27.         //声明队列
  28.         //参数1：消息队列名称
  29.         //参数2，是否持久化
  30.         //参数3，是否独占本次connection链接
  31.         //参数4，是否自动删除消息
  32.         //参数5，是否需要传递额外的队列数据Map
  33.         channel.queueDeclare("fanout_queue2",true,false,false,null);
  34.         //创建消费者，并且设置消息处理
  35.         //指定频道
  36.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
  37.             /**
  38.              *
  39.              * @param consumerTag 消费者名称，如果之前没有指定，则会自动生成一个
  40.              * @param envelope  消息一些额外的参数 channel.basicPublish("","simple_queue1",null,msg.getBytes());
  41.              * @param properties  另外的数据所封装的属性对象 null
  42.              * @param body  消息本身
  43.              * @throws IOException
  44.              */
  45.             @Override
  46.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  47.                 //实现消费的业务逻辑
  48.                 System.out.println("consumerTag"+consumerTag);
  49.                 System.out.println("交换机："+envelope.getExchange()+"----RoutingKey："+envelope.getRoutingKey()+"----DeliveryTag："+envelope.getDeliveryTag());
  50.                 System.out.println("消息本身："+new String(body,"UTF-8"));
  51.             }
  52.         };
  53.         //监听消息
  54.         //参数1，指定要监听的队列的名称
  55.         //参数2，设置消息的应答模式，true自动应答（性能好，消息有可能丢失），false手动应答
  56.         //参数3，
  57.         channel.basicConsume("fanout_queue2",true,defaultConsumer);
  58.  
  59.  
  60.         //关闭资源，建议不关闭一直监听
  61.     }
  62. }

订阅-Routing路由模式：

生产者代码

  1. package com.example.routingkey;
  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. import java.nio.charset.StandardCharsets;
  10. import java.util.concurrent.TimeoutException;
  11.  
  12. public class RoutingKeyProvider {
  13.     public static void main(String[] args) throws IOException, TimeoutException {
  14.         //创建链接工厂对象
  15.         ConnectionFactory connectionFactory = new ConnectionFactory();
  16.         //设置RabbitMQ服务主机地址，默认localhost
  17.         connectionFactory.setHost("localhost");
  18.         //设置端口，默认5672
  19.         connectionFactory.setPort(5672);
  20.         //设置虚拟主机名，默认/
  21.         connectionFactory.setVirtualHost("/mytest");
  22.         //设置用户名密码
  23.         connectionFactory.setUsername("test");
  24.         connectionFactory.setPassword("123456");
  25.  
  26.         //创建链接
  27.         Connection connection = connectionFactory.newConnection();
  28.         //创建通道
  29.         Channel channel = connection.createChannel();
  30.         //声明队列
  31.         //参数1：消息队列名称
  32.         //参数2，是否持久化
  33.         //参数3，是否独占本次connection链接
  34.         //参数4，是否自动删除消息
  35.         //参数5，是否需要传递额外的队列数据Map
  36.         channel.queueDeclare("routingkey_queue1",true,false,false,null);
  37.         //声明两个队列
  38.         channel.queueDeclare("routingkey_queue2",true,false,false,null);
  39.  
  40.         //声明交换机
  41.         //参数1 交换机名称
  42.         //参数2，指定交换机的类型 BuiltinExchangeType.FANOUT（广播） BuiltinExchangeType.DIRECT(路由模式)
  43.         channel.exchangeDeclare("routingkey_exchange", BuiltinExchangeType.DIRECT);
  44.         //将队列绑定到指定的交换机
  45.         //参数1，指定要绑定的队列
  46.         //参数2，指定绑定的交换机
  47.         //参数3，指定routingkey，广播模式可以不写, routingkey模式要写
  48.         channel.queueBind("routingkey_queue1","routingkey_exchange","item.insert");
  49.         channel.queueBind("routingkey_queue2","routingkey_exchange","item.delete");
  50.  
  51.         //创建消息
  52.         String msg = "增加操作";
  53.         //发布消息
  54.         //参数1，使用指定的交换机，不设置则使用默认交换机
  55.         //参数2，指定路由器key，如果是简单模式直接给队列名就行，广播模式指定为空
  56.         //参数3，发生消息时是否需要额外的消息数据
  57.         //参数4，消息内容
  58.         channel.basicPublish("routingkey_exchange","item.insert",null,msg.getBytes(StandardCharsets.UTF_8));
  59.         String msg2 = "删除操作";
  60.         //发布消息
  61.         //参数1，使用指定的交换机，不设置则使用默认交换机
  62.         //参数2，指定路由器key，如果是简单模式直接给队列名就行，广播模式指定为空
  63.         //参数3，发生消息时是否需要额外的消息数据
  64.         //参数4，消息内容
  65.         channel.basicPublish("routingkey_exchange","item.delete",null,msg2.getBytes(StandardCharsets.UTF_8));
  66.         //关闭资源
  67.         channel.close();
  68.         connection.close();
  69.     }
  70. }

消费者代码还跟上面一样，改一下对应的队列名就行

订阅-Topics通配符模式(主题模式)

RoutingKey一般是由一个或者多个单词组成，多个单词之间以.分割。例如：item.insert

通配符规则：

#：匹配一个或者多个词

*：匹配不多不少一个词

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

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

  

闽ICP备14008679号