/**
 * @Description RabbitMQ工具類
 * @date 2022/3/5 10:02
 */
public class RabbitMQUtils {
    public static Channel getChannel() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("1");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        return connection.createChannel();
    }
}

public class Worker1 {
    // 指定隊(duì)列名稱
    private static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {

        // 獲取信道
        Channel channel = RabbitMQUtils.getChannel();

        // 聲明：接收消息回調(diào)
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("工作線程01："+ new String(message.getBody()));
        };

        // 聲明：取消消費(fèi)回調(diào)
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("工作線程01取消接收："+consumerTag);
        };

        System.out.println("工作線程01啟動(dòng)完成......");

        channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);
    }
}

public class Producer {

    private static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMQUtils.getChannel();


        // 產(chǎn)生隊(duì)列
        channel.queueDeclare(QUEUE_NAME,false,false,true,null);

        // 消息體
        Scanner scanner = new Scanner(System.in);
        int i = 1;
        while (scanner.hasNext()){
            String msg = scanner.next();
            msg = msg + i;
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            System.out.println("發(fā)送成功：" + msg);
        }

        System.out.println("----------==========發(fā)送完畢==========----------");
    }

}

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue 'work_queue' in vhost '/', class-id=60, method-id=20)

/**
 * @Description 手動(dòng)應(yīng)答生產(chǎn)者
 * @date 2022/3/5 19:03
 */
public class Producer1 {

    // 指定隊(duì)列名
    private static final String TASK_QUEUE_RES = "queue_res";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        channel.queueDeclare(TASK_QUEUE_RES,false,false,false,null);
        Scanner scanner = new Scanner(System.in);
        int i = 0;
        while (scanner.hasNext()){
            i++;
            String msg = scanner.next() + i;
            channel.basicPublish("",TASK_QUEUE_RES,null,msg.getBytes(StandardCharsets.UTF_8));
            System.out.println("發(fā)送消息：'" + msg + "'成功");
        }
    }
}

/**
 * @Description 手動(dòng)應(yīng)答消費(fèi)者1
 * @date 2022/3/5 19:17
 */
public class Worker1 {

    private static final String TASK_QUEUE_RES = "queue_res";

    public static void main(String[] args)  throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        System.out.println("線程A等待接收......");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            // 模擬并發(fā)沉睡一秒
            try {
                Thread.sleep(1000);
                System.out.println("線程A接收消息："+ new String(message.getBody(), StandardCharsets.UTF_8));
                /**
                 * basicAck:
                 *          1. 消息標(biāo)記
                 *          2. 是否批量
                 */
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        channel.basicConsume(TASK_QUEUE_RES,false,deliverCallback,
                consumerTag -> {
                    System.out.println(consumerTag + "消費(fèi)者取消消費(fèi)");
                });

    }
}