【RabbitMQ】 Java简单的实现RabbitMQ

2022-05-10 07:07:43

准备工作

　　2、新建Java项目，引入RabbitMQ的Maven依赖

 <dependency>

     <groupId>com.rabbitmq</groupId>

     <artifactId>amqp-client</artifactId>

     <version>5.5.0</version>

 </dependency>

单生产者消费者

　　流程图：

　　1、创建生产者Producer1

 public class Producer1 {

     private final static String QUEUE_NAME = "rabbitMQ.test.queue";

     public static void main(String[] args) throws IOException, TimeoutException {

         // 创建连接工厂

         ConnectionFactory factory = new ConnectionFactory();

         // 设置RabbitMQ相关信息

         factory.setHost("localhost");

         factory.setUsername("guest");

         factory.setPassword("guest");

         factory.setPort(5672);

         // 创建一个新的连接

         Connection connection = factory.newConnection();

         // 创建一个通道

         Channel channel = connection.createChannel();

         // 声明一个队列

         // queueDeclare（队列名称，是否持久化（true表示是，队列将在服务器重启时生存），是否是独占队列（创建者可以使用的私有队列，断开后自动删除），

         //                        当所有消费者客户端连接断开时是否自动删除队列，队列的其他参数）

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         String message = "Hello RabbitMQ ～";

         // 发送消息到队列中

         // basicPublish（交换机名称，队列映射的路由key，消息的其他属性，发送信息的主体）

         channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));

          System.out.println("Producer Send +'" + message + "'");

         // 关闭通道和连接

         channel.close();

         connection.close();

     }

 }

　　2、创建消费者Customer1

 public class Customer1 {

     private final static String QUEUE_NAME = "rabbitMQ.test.queue";

     public static void main(String[] args) throws IOException, TimeoutException {

         // 创建连接工厂

         ConnectionFactory factory = new ConnectionFactory();

         // 设置RabbitMQ地址

         factory.setHost("localhost");

         factory.setUsername("guest");

         factory.setPassword("guest");

         factory.setPort(5672);

         // 创建一个新的连接

         Connection connection = factory.newConnection();

         // 创建一个通道

         final Channel channel = connection.createChannel();

         // 声明要关注的队列

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         channel.basicQos(1);//保证一次只分发一个

         System.out.println("Customer Waiting Received messages");

         // DefaultConsumer类实现了Consumer接口，通过传入一个频道，

         // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery

         Consumer consumer = new DefaultConsumer(channel) {

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,

                     byte[] body) throws IOException {

                 String message = new String(body, "UTF-8");

                 System.out.println("Customer Received '" + message + "'");

             }

         };

         // 自动回复队列应答 -- RabbitMQ中的消息确认机制

         channel.basicConsume(QUEUE_NAME, true, consumer);

     }

 }

　　3、运行结果

　　　　a、生产者

　　　　b、消费者

推送确认和消费应答

　　流程图

　　1、创建推送确认生产者Producer2

 public class Producer2 {

     private final static String QUEUE_NAME = "rabbitMQ.test.queue";

     public static void main(String[] args) throws IOException, TimeoutException {

         // 创建连接工厂

         ConnectionFactory factory = new ConnectionFactory();

         // 设置RabbitMQ相关信息

         factory.setHost("localhost");

         factory.setUsername("guest");

         factory.setPassword("guest");

         factory.setPort(5672);

         // 创建一个新的连接

         Connection connection = factory.newConnection();

         // 创建一个通道

         Channel channel = connection.createChannel();

         // 声明一个队列

         // queueDeclare（队列名称，是否持久化（true表示是，队列将在服务器重启时生存），是否是独占队列（创建者可以使用的私有队列，断开后自动删除），

         // 当所有消费者客户端连接断开时是否自动删除队列，队列的其他参数）

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         // 开启发送方确认模式

         channel.confirmSelect();

         // 存储未确认的消息标识tag

         final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());

         // 异步监听确认和未确认的消息

         channel.addConfirmListener(new ConfirmListener() {

             /**

              * 处理返回确认成功

              *

              * @param deliveryTag 如果是多条，这个就是最后一条消息的tag

              * @param multiple    是否多条

              * @throws IOException

              */

             public void handleAck(long deliveryTag, boolean multiple) throws IOException {

                 System.out.println("消息发送成功,    deliveryTag:" + deliveryTag + "        multiple:" + multiple + "");

                 if (multiple) {

                     // 移除发送成功的多条消息标识tag

                     confirmSet.headSet(deliveryTag + 1).clear();

                 } else {

                     // 移除发送成功的一条消息标识tag

                     confirmSet.remove(deliveryTag);

                 }

             }

             /**

              * 处理返回确认失败

              *

              * @param deliveryTag 如果是多条，这个就是最后一条消息的tag

              * @param multiple    是否多条

              * @throws IOException

              */

             public void handleNack(long deliveryTag, boolean multiple) throws IOException {

                 System.out.println("失败,deliveryTag:" + deliveryTag + "multiple:" + multiple + "");

                 if (multiple) {

                     confirmSet.headSet(deliveryTag + 1).clear();

                 } else {

                     confirmSet.remove(deliveryTag);

                 }

             }

         });

         String message = "Hello RabbitMQ ～ ";

         // 发送消息到队列中

         // basicPublish（交换机名称，队列映射的路由key，消息的其他属性，发送信息的主体）

         for (int i = 1; i <= 10; i++) {

             String msg = message + i;

             long tag = channel.getNextPublishSeqNo();

             confirmSet.add(tag);

             System.out.println("tag:" + tag);

             channel.basicPublish("", QUEUE_NAME, null, msg.getBytes("UTF-8"));

             System.out.println("Producer Send +'" + msg + "'");

         }

         System.out.println("============================");

         // 关闭通道和连接

         // channel.close();

         // connection.close();

     }

 }

　　2、创建消费应答消费者Customer2

 public class Customer2 {

     private final static String QUEUE_NAME = "rabbitMQ.test.queue";

     public static void main(String[] args) throws IOException, TimeoutException {

         // 创建连接工厂

         ConnectionFactory factory = new ConnectionFactory();

         // 设置RabbitMQ地址

         factory.setHost("localhost");

         factory.setUsername("guest");

         factory.setPassword("guest");

         factory.setPort(5672);

         // 创建一个新的连接

         Connection connection = factory.newConnection();

         // 创建一个通道

         final Channel channel = connection.createChannel();

         // 声明要关注的队列

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         channel.basicQos(1);// 保证一次只分发一个

         System.out.println("Customer Waiting Received messages");

         // DefaultConsumer类实现了Consumer接口，通过传入一个频道，

         // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery

         Consumer consumer = new DefaultConsumer(channel) {

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,

                     byte[] body) throws IOException {

                 String message = new String(body, "UTF-8");

                 System.out.println("Customer Received '" + message + "'");

                 // 返回消费确认状态

                 channel.basicAck(envelope.getDeliveryTag(), false);

             }

         };

         // 消费手动确认 -- RabbitMQ中的消息确认机制

         channel.basicConsume(QUEUE_NAME, false, consumer);

     }

 }

　　3、运行结果

　　　　生产者：
　　　　　　

　　　　消费者：
　　　　　　

多消费者

　　流程图
　　　　

　　1、创建推送确认生产者Producer3（与示例Producer2一样）

　　2、创建消费应答消费者Customer3

 public class Customer3 {

     private final static String QUEUE_NAME = "rabbitMQ.test.queue";

     public static void main(String[] args) throws IOException, TimeoutException {

         Customer3 customer3 = new Customer3();

         customer3.createCustomer("customer1");

         customer3.createCustomer("customer2");

     }

     public void createCustomer(final String customerName) throws IOException, TimeoutException{

         // 创建连接工厂

         ConnectionFactory factory = new ConnectionFactory();

         // 设置RabbitMQ地址

         factory.setHost("localhost");

         factory.setUsername("guest");

         factory.setPassword("guest");

         factory.setPort(5672);

         // 创建一个新的连接

         Connection connection = factory.newConnection();

         // 创建一个通道

         final Channel channel = connection.createChannel();

         // 声明要关注的队列

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         channel.basicQos(1);// 保证一次只分发一个

         System.out.println(customerName + " Waiting Received messages");

         // DefaultConsumer类实现了Consumer接口，通过传入一个频道，

         // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery

         Consumer consumer = new DefaultConsumer(channel) {

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,

                     byte[] body) throws IOException {

                 String message = new String(body, "UTF-8");

                 System.out.println(customerName + " Received '" + message + "'");

                 // doWork处理任务

                 doWork(customerName);

                 // 返回消费确认状态

                 channel.basicAck(envelope.getDeliveryTag(), false);

             }

         };

         // 消费手动确认 -- RabbitMQ中的消息确认机制

         channel.basicConsume(QUEUE_NAME, false, consumer);

     }

     private void doWork(String customer) {

         try {

             Thread.sleep(2000); // 暂停2秒钟

             System.out.println(customer + ": completion of the job!");

         } catch (InterruptedException _ignored) {

             Thread.currentThread().interrupt();

         }

     }

 }

　　3、运行结果
　　　　生产者与前面相同

　　　　消费者
　　　　　　
　　

码农公寓

准备工作

单生产者消费者

推送确认和消费应答

多消费者

相关文章