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);

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

    }

}