RabbitMQ整合Spring Booot【生產者事務確認機制】

時間 2019-11-06

標籤 rabbitmq 整合 spring booot 生產者事務確認機制欄目 RabbitMQ 简体版

原文原文鏈接

package com.toov5.amqp;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.toov5.utils.MQConnectionUtils;

public class Producer {
    // 隊列名稱
    private static final String UEUE_NAME = "test_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 建立新的鏈接
        Connection connection = MQConnectionUtils.newConnection();
        // 建立Channel
        Channel channel = connection.createChannel();
        // 建立隊列
        channel.queueDeclare(UEUE_NAME, false, false, false, null);
        channel.basicQos(1); // 保證 取一個消費
            try {
               channel.txSelect(); //開啓事務 // 建立message
                String msg = "toov5_message";
                System.out.println("生產者投遞消息" + msg );
                // 生產者發送消息
                channel.basicPublish("", UEUE_NAME, null, msg.getBytes());
                int i = 1/0;
               channel.txCommit(); //提交事務
            } catch (Exception e) {
                System.out.println("生產者消息事務已經回滾");
              channel.txRollback(); //回滾事務
            }finally {
                // 關閉通道和鏈接
                channel.close();
                connection.close();

            }            
            
        }
    
    }

consumer:java

package com.toov5.amqp;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.toov5.utils.MQConnectionUtils;

public class Consumer1 {
  
     //隊列名稱
        private static final String QUEUE_NAME = "test_queue";
        
        public static void main(String[] args) throws IOException, TimeoutException {
            System.out.println("消費者啓動..........1");
            //建立新的鏈接
        Connection connection = MQConnectionUtils.newConnection();
           //建立Channel
            final Channel channel = connection.createChannel();
            // 消費者關聯隊列
             channel.queueDeclare(QUEUE_NAME, false, false, false, null);
             channel.basicQos(1);
              DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) {
                  //監聽獲取消息
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                            byte[] body) throws IOException {
                        String msg =new String(body,"UTF-8");
                        System.out.println("消費者獲取生產者消息："+msg);
                        try {
                            //模擬應答等待時間
                            Thread.sleep(1000);
                        } catch (Exception e) {
                            
                        }finally {
                           channel.basicAck(envelope.getDeliveryTag(), false);  //手動應答 告訴消息隊列服務器 消費成功
                        }
                    }
              };
            //牽手模式設置  默認自動應答模式  true:自動應答模式  
              channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);//    fanse手動應答          

        }
}