package com.xiaojie.rabbitmq.tx;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xiaojie.rabbitmq.MyConnection;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
/**
 * @Description: mq事务模式保证消息可靠性
 * @author: xiaojie
 * @date: 2021.09.28
 */
public class TxProvider {
    //定义队列
    private static final String QUEUE_NAME = "myqueue";
    static Channel channel = null;
    static Connection connection = null;
    public static void main(String[] args) {
        try {
            System.out.println("生产者启动成功..");
            // 1.创建连接
            connection = MyConnection.getConnection();
            // 2.创建通道
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String msg = "测试事务机制保证消息发送可靠性。。。。";
            channel.txSelect(); //开启事务
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes(StandardCharsets.UTF_8));
            //发生异常时，mq中并没有新的消息入队列
            //int i=1/0;
            //没有发生异常，提交事务
            channel.txCommit();
            System.out.println("生产者发送消息成功:" + msg);
        } catch (Exception e) {
            e.printStackTrace();
            //发生异常则回滚事务
            try {
                if (channel != null) {
                    channel.txRollback();
                }
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        } finally {
            try {
                if (channel != null) {
                    channel.close();
                }
                if (connection != null) {
                    connection.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }
}

package com.xiaojie.rabbitmq.tx;
import com.rabbitmq.client.*;
import com.xiaojie.rabbitmq.MyConnection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @Description: 事务模拟消费者
 * 如果不改为手动应达模式，那么事务开启对消费者没有影响
 * @author: xiaojie
 * @date: 2021.09.28
 */
public class Txconsumer {
    private static final String QUEUE_NAME = "myqueue";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.创建我们的连接
        Connection connection = MyConnection.getConnection();
        // 2.创建我们通道
        Channel channel = connection.createChannel();
        channel.txSelect();//开启事务
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");
                //设置手动应答
                channel.basicAck(envelope.getDeliveryTag(), true);
                System.out.println("消费消息msg:" + msg+"手动应答："+envelope.getDeliveryTag());
                channel.txCommit(); //消费者开启事务，必须要提交事务之后，消息才会从队列中移除，否则不移除。
            }
        };
        // 3.创建我们的监听的消息 false 关闭自动确认模式
        channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    }
}

//队列持久化
 channel.queueDeclare(EMAIL_QUEUE_FANOUT, true, false, false, null);
 //生产者绑定交换机 参数1 交换机的名称。2，交换机的类型,3,true标识持久化
 channel.exchangeDeclare(EXCHANGE, "direct",true);

package com.xiaojie.springboot.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
/**
 * @author xiaojie
 * @version 1.0
 * @description: 配置rabbitmq
 * @date 2021/9/25 22:16
 */
@Component
public class RabbitMqConfig {
    //定义队列
    private static final String MY_FANOUT_QUEUE = "xiaojie_fanout_queue";
    //定义队列
    private static final String MY_DIRECT_QUEUE = "xiaojie_direct_queue";
    //定义队列
    private static final String MY_TOPIC_QUEUE = "xiaojie_topic_queue";
    //定义fanout交换机
    private static final String MY_FANOUT_EXCHANGE = "xiaojie_fanout_exchange";
    //定义direct交换机
    private static final String MY_DIRECT_EXCHANGE = "xiaojie_direct_exchange";
    //定义topics交换机
    private static final String MY_TOPICS_EXCHANGE = "xiaojie_topics_exchange";
    //创建队列 默认开启持久化
    @Bean
    public Queue fanoutQueue() {
        return new Queue(MY_FANOUT_QUEUE);
    }
    //创建队列
    @Bean
    public Queue directQueue() {
        return new Queue(MY_DIRECT_QUEUE);
    }
    //创建队列
    @Bean
    public Queue topicQueue() {
        return new Queue(MY_TOPIC_QUEUE);
    }
    //创建fanout交换机 默认开启持久化
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(MY_FANOUT_EXCHANGE);
    }
    //创建direct交换机
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(MY_DIRECT_EXCHANGE);
    }
    //创建direct交换机
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(MY_TOPICS_EXCHANGE);
    }
    //绑定fanout交换机
    @Bean
    public Binding fanoutBindingExchange(Queue fanoutQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue).to(fanoutExchange);
    }
    //绑定direct交换机
    @Bean
    public Binding directBindingExchange(Queue directQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue).to(directExchange).with("msg.send");
    }
    //绑定topic交换机
    @Bean
    public Binding topicBindingExchange(Queue topicQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicQueue).to(topicExchange).with("msg.#");
    }
    //多个队列绑定到同一个交换机
//    @Bean
//    public Binding topicBindingExchange1(Queue directQueue, TopicExchange topicExchange) {
//        return BindingBuilder.bind(directQueue).to(topicExchange).with("msg.send");
//    }
}

package com.xiaojie.springboot.provider;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
/**
 * @author xiaojie
 * @version 1.0
 * @description:消息生产者 publisher-confirm-type: correlated
 * #NONE值是禁用发布确认模式，是默认值
 * #CORRELATED值是发布消息成功到交换器后会触发回调方法
 * #SIMPLE值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，
 * 其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，
 * 根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，将无法发送消息到broker
 * @date 2021/9/25 22:47
 */
@Component
@Slf4j
public class MsgProvider {
    //定义队列
    private static final String MY_FANOUT_QUEUE = "xiaojie_fanout_queue";
    private static final String MY_DIRECT_QUEUE = "xiaojie_direct_queue";
    private static final String DIRECT_ROUTING_KEY = "msg.send";
    private static final String MY_TOPIC_QUEUE = "xiaojie_topic_queue";
    private static final String TOPIC_ROUTING_KEY = "msg.send";
    //定义fanout交换机
    private static final String MY_FANOUT_EXCHANGE = "xiaojie_fanout_exchange";
    //定义direct交换机
    private static final String MY_DIRECT_EXCHANGE = "xiaojie_direct_exchange";
    //定义topics交换机
    private static final String MY_TOPICS_EXCHANGE = "xiaojie_topics_exchange";
    @Autowired
    private AmqpTemplate amqpTemplate;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public String sendMSg(Integer type, String msg) {
        try {
            if (1 == type) {
                //发送到fanout
                rabbitTemplate.convertAndSend(MY_FANOUT_QUEUE, "fanout" + msg);
            } else if (2 == type) {
                //第一个参数交换机，第二个参数，路由键，第三个参数，消息
                rabbitTemplate.convertAndSend(MY_DIRECT_EXCHANGE, DIRECT_ROUTING_KEY, "direct" + msg);
            } else if (3 == type) {
                rabbitTemplate.convertAndSend(MY_TOPICS_EXCHANGE, TOPIC_ROUTING_KEY, "topic" + msg);
            }
            return "success";
        } catch (AmqpException e) {
            e.printStackTrace();
        }
        return "error";
    }
}