package com.windlinxy.pqc;

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

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author Windlinxy
 * @description 生产者
 * @date 2022-06-28 15:45
 **/
public class Producer {
    // 队列名称
    public static final String QUEUE_NAME = "hello";


    /**
     * 发消息
     *
     * @param args 参数
     */
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 工厂ip连接RabbitMQ的队列
        factory.setHost("服务器ip");
        factory.setUsername("admin");
        factory.setPassword("123");

        // 创建连接
        Connection connection = factory.newConnection();
        // 连接中创建信道（具体看原理图）
        Channel channel = connection.createChannel();
        // 简单连接，忽略交换机，连接队列
        /*
         * 生成一个队列
         * 1. 队列名称
         * 2. 队列里的消息是否持久化（磁盘） 默认情况是存到内存（false）、
         * 3. 该队列是否只供一个消费者进行消费，是否消息共享，默认是false
         * 4. 最后一个消费者断开了连接之后，是否自动删除
         * */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 准备发消息
        String message = "hello world!";
/*       * 发送
         * 1. 发送到哪个交换机
         * 2. 路由的Key值是哪个，本次是队列的名称
         * 3. 其他参数信息
         * 4。 发送消息体 */
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));

        System.out.println("Success!");

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

package com.windlinxy.pqc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Arrays;

/**
 * @author Windlinxy
 * @description 消费者
 * @date 2022-06-28 15:45
 **/
public class Consumer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("服务器ip");
        factory.setUsername("admin");
        factory.setPassword("123");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        System.out.println("等待接收消息.........");
        //推送的消息如何进行消费的接口回调
        com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
            /*
            回调方法
            1. consumerTag 标识
            2. envelope 获取一些信息，交换机，路由key
            3. properties 配置信息
            4. body 数据
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag ：" + consumerTag);
                System.out.println("Exchange ： "+envelope.getExchange());
                System.out.println("RoutingKey ： "+envelope.getRoutingKey());
//                System.out.println("properties ： " + properties);
                System.out.println("body ： " + new String(body));
            }
        };

        /*
        高级用法
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println(message);
        };
        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("消息消费被中断");
        };
        */
        
        /*
         * 消费者消费消息
         * 1.消费哪个队列
         * 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
         * 3.消费者未成功消费的回调
         */
        channel.basicConsume(QUEUE_NAME, true, consumer);

        //消费者是监听消息，不要去关闭资源
    }
}