RabbitMQ-hello

0. pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.hong.rabbitmq</groupId>
    <artifactId>rabbitmq-hello</artifactId>
    <version>1.0-SNAPSHOT</version>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <encoding>utf-8</encoding>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.8.0</version>
        </dependency>

        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.6</version>
        </dependency>
    </dependencies>
</project>

1. 生产者代码

package com.hong.rabbitmq1;

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

public class Producer {
    //队列名称
    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("10.211.55.4");
        factory.setUsername("admin");
        factory.setPassword("123456");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        /**
         * 声明消息队列
         * 第一个参数: 队列名称
         * 第二个参数: 是否持久化     false不持久化,rabbitmq重启队列丢失  true持久化,rabbitmq重启队列不丢失
         * 第三个参数: 是否排外       false排外只能一个消费者消费         false不排外可多消费者消费
         * 第四个参数: 是否自动删除   true自动删除最后一个消费者断开连接后该队列自动删除   false不自动删除
         * 第五个参数: 其他参数信息
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        String message = "HELLO RABBITMQ!";
        /**
         * 发布消息
         * 第一个参数:要将消息发送到哪个交换机
         * 第二个参数:路由的 key 是哪个
         * 第三个参数:其他参数信息
         * 第四个参数:消息的消息体
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("消息发送完成!");

    }

}

2.消费者代码

package com.hong.rabbitmq1;

import com.rabbitmq.client.*;

public class Consumer {
    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("10.211.55.4");
        factory.setUsername("admin");
        factory.setPassword("123456");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        DeliverCallback deliverCallback = (comsumerTag,message) -> System.out.println(new String(message.getBody()));
        CancelCallback cancelCallback = var -> System.out.println(var + "消息消费被中断!");
        /**
         * 消息消费
         * 第一个参数: 消费哪个队列
         * 第二个参数: 是否自动确认消息
         * 第三个参数: 当一个消息发送过来后的回调接口
         * 第四个参数: 当一个消费者取消订阅时的回调接口;取消消费者订阅队列时除了使用{@link Channel#basicCancel}之外的所有方式都会调用该回调方法
         */
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

3.抽取工具类

package com.hong.utils;

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

/**
 * RabbitMQ创建Channel工具类
 */
public class RabbitMQUtil {
    public static Channel getChannel() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("10.211.55.4");
        factory.setUsername("admin");
        factory.setPassword("123456");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

4.工作队列

4.1.生产者代码

package com.hong.rabbitmq2;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;

import java.util.Scanner;

public class Task {
    //队列名称
    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        Scanner scanner = new Scanner(System.in);
        System.out.println("请输入:");
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("消息发送完成------" + message);
        }
    }
}

4.2.消费者代码

package com.hong.rabbitmq2;

import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

public class Worker {
    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        DeliverCallback deliverCallback = (comsumerTag, message) -> System.out.println("接收到的消息:"+  new String(message.getBody()));
        CancelCallback cancelCallback = var -> System.out.println(var + "消息消费被中断!");
        System.out.println("worker1等待接收消息");
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

4.3.说明

Worker启动2个worker1,worker2;两者依次消费消息(轮训/轮询)