package ppppp.demo;
import com.rabbitmq.client.*;
import org.junit.Test;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author lppppp
 * @create 2021-05-21 14:58
 */
public class RabbitProducer_Consumer {

    private static final String EXCHANGE_NAME="exchange-kk";
    private static final String ROUTING_KEY="routingkey-kk";
    private static final String QUEUE_NAME="queue-kk";
    private static final String USER_NAME="guest";
    private static final String PASSWORD="guest";
    private static final String IP_ADDRESS="127.0.0.1";
    private static final int PORT=5672; //默认端口号


    @Test
    public void producer() throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        // 获取新的连接
        Connection connection = factory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        // 创建一个 type="direct"、持久化的、非自动删除的交换器
        channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,null);
        // 创建一个持久化、非排他的、非自动删除的交换器
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        // 将交换器与队列通过路由键绑定
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
        String message="hello world";
        // 发送一条持久化的消息
        channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
        // 关闭信道
        channel.close();
        // 关闭资源
        connection.close();
    }

    @Test
    public void consumer() throws IOException, TimeoutException, InterruptedException {
        Address[] addresses = {new Address(IP_ADDRESS, PORT)};
        ConnectionFactory factory=new ConnectionFactory();
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        Connection connection = factory.newConnection(addresses);
        final Channel channel = connection.createChannel();
        // 设置客户端最多接收未被ack的消息的个数
        channel.basicQos(64);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 客户端接收后如何处理消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("rev message:"+new String(body));
                // 确认收到消息
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(QUEUE_NAME,consumer);
        // 等到回调函数完成后关闭连接
        TimeUnit.SECONDS.sleep(2);
        channel.close();
        connection.close();
    }

}

package com.qfedu.send;

import com.qfedu.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Send {

    public static void main(String[] argv) throws Exception {
        // 获取到连接以及mq通道
        Connection connection = ConnectionUtil.getConnection();
        // 相当于数据库中的创建连接
        // 从连接中创建通道
        Channel channel = connection.createChannel();
        // 相当于数据库中的 statement
        // 声明（创建）队列,如果存在就不创建,不存在就创建
        // 参数1 队列名,
        // 参数2 durable： 是否持久化, 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失，如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库
        // 参数3 exclusive：是否排外的，有两个作用，一：当连接关闭时connection.close()该队列是否会自动删除； 二：该队列是否是私有的private，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外 的，会对当前队列加锁，其他通道channel是不能访问的，如果强制访问会报异常： com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=405, reply-text=RESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'queue_name' in vhost '/', class-id=50, method-id=20)一般等于true的话 用于一个队列只能有一个消费者来消费的场景
        // 参数4 autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除，可以通过RabbitMQ Management，查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
        // 参数5 arguments：  参数
        //channel.queueDeclare("queue1", false, false, true, null);
        // 消息内容
         String message = "Hello World!";
        // 参数1 交换机,此处无
        // 参数2 发送到哪个队列
        // 参数3 属性
        // 参数4 内容
         channel.basicPublish("", "queue1", null, message.getBytes());
        // 将消息发动到数据库
         System.out.println(" 发送数据 '" + message + "'");
        //关闭通道和连接
         channel.close();
         connection.close();
    }
}

package com.qfedu.receive;

import com.qfedu.utils.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Receive {

    public static void main(String[] argv) throws Exception {
        //创建一个新的连接
        Connection connection = ConnectionUtil.getConnection();
        //创建一个通道
        Channel channel = connection.createChannel();

        //声明要关注的队列
        //channel.queueDeclare("queue1", false, false, false, null);
        //DefaultConsumer类实现了Consumer接口，通过传入一个频道，
        // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
        Consumer consumer = new DefaultConsumer(channel) {
            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("queue1", true, consumer);
    }
}

package ppppp.producer.controller;

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import ppppp.producer.service.MQService;

import javax.annotation.Resource;

@RestController
public class DemoCtroller {
    @Resource
    private MQService mqService;

    @RequestMapping("test")
    public String demo(String msg){
        mqService.sendMsg(msg);
        return "success";
    }
}

package ppppp.producer.service;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class MQService {

    @Resource
    private  AmqpTemplate amqpTemplate;
    public void sendMsg(String msg) {
        if(msg.startsWith("q_")){
            //1. 发送消息到队列
            amqpTemplate.convertAndSend("queue-kk",msg);
        }else if(msg.startsWith("f_")){
            //2. 发送消息到订阅交换机
            amqpTemplate.convertAndSend("ex1","",msg);
        }else if(msg.startsWith("r_")){
            //3. 发送消息到路由交换机
            if(msg.startsWith("r_a")){
                amqpTemplate.convertAndSend("ex2","a",msg);
            }else if(msg.startsWith("r_b")){
                amqpTemplate.convertAndSend("ex2","b",msg);
            }
        }
    }
}

// 1.生产者
package ppppp.mq.service;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.SerializationUtils;
import ppppp.mq.bean.Goods;

@Service
public class MQService {
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendMsg(Goods goods){
        byte[] serializeGoods = SerializationUtils.serialize(goods);
        amqpTemplate.convertAndSend("","queue-kk",serializeGoods);
    }
}

//2.消费者
@RabbitHandler
    public void receiveMQ(byte[] msg){
        //System.out.println("byte[] ---  " + msg);
        Goods goods = (Goods) SerializationUtils.deserialize(msg);
        System.out.println("byte[] ---  " + goods);
    }

//1.开启事务
channel.txSelect();

//2.提交事务
channel.txCommit();

//3.事务回滚
channel.txRollback();

Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
String message = "Hello World!";

//1.开启消息确认
channel.confirmSelect();
//2.发送消息
channel.basicPublish("ex2", "c", null, message.getBytes());
//3.获取确认
boolean b = channel.waitForConfirms();
System.out.println("消息发送"+(b?"成功":"失败"));

channel.close();
connection.close();

Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
String message = "Hello World!";

//1.开启消息确认
channel.confirmSelect();
//2.发送消息
for (int i=1; i<=10; i++) {
    message += i;
    channel.basicPublish("ex2", "c", null, message.getBytes());
}
//3.批量确认:发送的所有消息中有一个失败就直接全部失败，抛出IO异常
boolean b = channel.waitForConfirms();

channel.close();
connection.close();

Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
String message = "Hello World!";

//1.开启消息确认
channel.confirmSelect();
//2.发送消息
for (int i=1; i<=10; i++) {
    message += i;
    channel.basicPublish("ex2", "c", null, message.getBytes());
}

//3.开启异步confirm
channel.addConfirmListener(new ConfirmListener() {
    //参数l表示返回的消息标识，参数b表示是否为批量confirm
    public void handleAck(long l, boolean b) throws IOException {
        System.out.println("----消息发送成功");
    }
    public void handleNack(long l, boolean b) throws IOException {
        System.out.println("----消息发送失败");
    }
});

channel.close();
connection.close();

@Test
public void T_异步批量发送消息() throws Exception {
    Connection connection = getConnection();
    Channel channel = connection.createChannel();
    String message = "Hello World!";

    //1.开启消息确认
    channel.confirmSelect();

    //3.开启异步confirm
    channel.addConfirmListener(new ConfirmListener() {
        //参数l表示返回的消息标识，
        //参数b表示是否为批量confirm
        public void handleAck(long l, boolean b) throws IOException {
            System.out.println("----消息发送成功");
        }
        public void handleNack(long l, boolean b) throws IOException {
            System.out.println("----消息发送失败");
        }
    });

    //4.添加return监控
    channel.addReturnListener(new ReturnListener() {
        //消息未分发到队列中时，会执行此语句
        @Override
        public void handleReturn(int replyCode,
                                 String replyText,
                                 String exchange,
                                 String routingKey,
                                 AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.out.println("发送失败*****");
            System.out.println("replyCode————"+replyCode);
            System.out.println("replyText————"+replyText);
            System.out.println("exchange————"+exchange);
            System.out.println("routingKey————"+routingKey);
            System.out.println("body————"+new String(body));
        }
    });

    //2.发送消息
    for (int i=1; i<=10; i++) {
        message += i;
        //channel.basicPublish("ex2", "c", null, message.getBytes());
        channel.basicPublish("ex2", "c",true, null, message.getBytes());
    }
    channel.close();
    connection.close();
}

package com.qfedu.mq_producer.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@Component
public class PublisherConfireAndReturnConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    Logger logger = LoggerFactory.getLogger(PublisherConfireAndReturnConfig.class);

    @Resource
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void initMethod(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        if(ack){
            logger.info("--------消息发送(到交换机)成功");
        }else{
            logger.warn("--------消息发送(到交换机)失败");
        }
    }

    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {
        logger.info("~~~~~~~~消息发送到交换机但未分发到队列！！！");
    }
}

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
    .deliveryMode(1)     //指定消息书否需要持久化 1 - 需要持久化  2 - 不需要持久化
    .messageId(UUID.randomUUID().toString())
    .build();
String msg = "Hello-World！";
channel.basicPublish("","HelloWorld",true,properties,msg.getBytes());

DefaultConsumer consume = new DefaultConsumer(channel){
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        Jedis jedis = new Jedis("192.168.199.109",6379);
        String messageId = properties.getMessageId();
        //1. setnx到Redis中，默认指定value-0
        String result = jedis.set(messageId, "0", "NX", "EX", 10);
        if(result != null && result.equalsIgnoreCase("OK")) {
            System.out.println("接收到消息：" + new String(body, "UTF-8"));
            //2. 消费成功，set messageId 1
            jedis.set(messageId,"1");
            channel.basicAck(envelope.getDeliveryTag(),false);
        }else {
            //3. 如果1中的setnx失败，获取key对应的value，如果是0，return，如果是1
            String s = jedis.get(messageId);
            if("1".equalsIgnoreCase(s)){
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        }
    }
};

@Test
void contextLoads() throws IOException {
    CorrelationData messageId = new CorrelationData(UUID.randomUUID().toString());
    rabbitTemplate.convertAndSend("boot-topic-exchange","slow.red.dog","红色大狼狗！！",messageId);
    System.in.read();
}

@Autowired
private StringRedisTemplate redisTemplate;


@RabbitListener(queues = "boot-queue")
public void getMessage(String msg, Channel channel, Message message) throws IOException {
    //0. 获取MessageId
    String messageId = message.getMessageProperties().getHeader("spring_returned_message_correlation");
    //1. 设置key到Redis
    if(redisTemplate.opsForValue().setIfAbsent(messageId,"0",10, TimeUnit.SECONDS)) {
        //2. 消费消息
        System.out.println("接收到消息：" + msg);

        //3. 设置key的value为1
        redisTemplate.opsForValue().set(messageId,"1",10,TimeUnit.SECONDS);
        //4.  手动ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }else {
        //5. 获取Redis中的value即可 如果是1，手动ack
        if("1".equalsIgnoreCase(redisTemplate.opsForValue().get(messageId))){
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
    }
}

package ppppp.mq;

import com.rabbitmq.client.*;

import java.io.IOException;

import static ppppp.mq.P2ApplicationTests.getConnection;

public class T_consumer {
    public static void main(String[] args) throws Exception {
        //创建一个新的连接
        Connection connection = getConnection();
        //创建一个通道
        Channel channel = connection.createChannel();

        // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
        Consumer consumer = new DefaultConsumer(channel) {
            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_delay2", true, consumer);
    }

}

package ppppp.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.util.Scanner;

import static ppppp.mq.P2ApplicationTests.getConnection;

public class T_producer {

    public static void main(String[] args) throws Exception {
        Scanner scanner = new Scanner(System.in);
        String  msg = null;
        while (!"q".equalsIgnoreCase(msg=scanner.nextLine())){
            Connection connection = getConnection();
            Channel channel = connection.createChannel();
            String message = "Hello World ————" + msg;

            channel.basicPublish("delay_exchange", "k1", null, message.getBytes());
            channel.close();
            connection.close();
        }
    }

}