基于 RabbitMQ 的消息发布与订阅

　pom.xml

了解springcloud架构可以加求求：三五三六二四七二五九

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

 RabbitMQ 连接配置工具类

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

public class ChannelUtils {

    public static Channel getChannelInstance(String connectionDescription) {
        try {
            ConnectionFactory connectionFactory = getConnectionFactory();
            // 创建连接
            Connection connection = connectionFactory.newConnection(connectionDescription);
            return connection.createChannel();
        } catch (Exception e) {
            throw new RuntimeException("获取Channel连接失败");
        }
    }

    /**
     * 功能描述:
     * <连接配置>
     *
     *
     * @return com.rabbitmq.client.ConnectionFactory
     * @author zhoulipu
     * @date   2019/7/31 18:02
     */
    private static ConnectionFactory getConnectionFactory() {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 配置连接信息
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        // 失败重连模式：网络异常自动重新连接
        connectionFactory.setAutomaticRecoveryEnabled(true);
        // 失败重连模式：每10秒重试连接
        connectionFactory.setNetworkRecoveryInterval(10000);
        // 失败重连模式：重新声明交换器，队列等信息
        connectionFactory.setTopologyRecoveryEnabled(true);
        return connectionFactory;
    }

}

 

生产者发布消息

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Scanner;

public class Producer {

    /**
     * 交换机名称（需与消费者保持一致）
     */
    private static final String pbxName = "PBX";

    /**
     * 路由密匙（需与消费者保持一致）
     */
    private static final String routingKey = "routingKey";

    public static void main(String[] args) throws IOException {
        // 获取连接
        Channel channel = ChannelUtils.getChannelInstance("队列消息生产者");
        // 生产者配置
        AMQP.BasicProperties basicProperties = producerConfig(channel, pbxName);
        while (true) {
            System.out.print("请输入即将发布的消息：");
            Scanner str = new Scanner(System.in);
            // 发布消息
            publish(channel, basicProperties, pbxName, routingKey, str.next());
            System.out.println("消息已发布，请查看订阅者状态");
        }
    }

    /**
     * 功能描述:
     * <生产者发布消息>
     *
     * @param channel         1
     * @param basicProperties 2
     * @param pbxName         3
     * @param routingKey      4
     * @param msg             5
     * @return void
     * @author zhoulipu
     * @date 2019/7/31 17:52
     */
    public static void publish(Channel channel, AMQP.BasicProperties basicProperties, String pbxName, String routingKey, String msg) throws IOException {
        // 发布消息(交换机名, 路由关键字, 是否为强制性, 消息属性, 消息体);
        channel.basicPublish(pbxName, routingKey, false, basicProperties, msg.getBytes());
    }

    /**
     * 功能描述:
     * <生产者配置>
     *
     * @param channel 1
     * @param pbxName 2
     * @return com.rabbitmq.client.AMQP.BasicProperties
     * @author zhoulipu
     * @date 2019/7/31 17:52
     */
    public static AMQP.BasicProperties producerConfig(Channel channel, String pbxName) throws IOException {
        // 声明交换机 (交换机名称, 交换机类型, 是否持久化, 是否自动删除, 是否是内部交换机（客户机直接发送消息）, 队列中的消息自动删除配置);
        channel.exchangeDeclare(pbxName, BuiltinExchangeType.DIRECT, true, false, false, new HashMap<>());
        // 消息属性
        AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder()
                .deliveryMode(2) // 设置消息是否持久化，1： 非持久化 2：持久化
                .contentType("UTF-8")
                .contentEncoding("UTF-8")
                .headers(new HashMap<>())
                .build();
        return basicProperties;
    }

}

 

消费者订阅消息

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class Consumer {

    /**
     * 交换机名称（需与消费者保持一致）
     */
    private static final String pbxName = "PBX";

    /**
     * 路由密匙（需与消费者保持一致）
     */
    private static final String routingKey = "routingKey";


    public static void main(String[] args) throws IOException {
        // 队列名称（多个消费者需使用不同队列名称）
        String queueName = "queueName";
        // 获取连接
        Channel channel = ChannelUtils.getChannelInstance("队列消息消费者");
        // 消费者配置（交换机、路由、队列）
        AMQP.Queue.DeclareOk declareOk = consumerConfig(channel, pbxName, routingKey, queueName);
        // 订阅消息
        subscribe(channel, declareOk, "消费者名称");
    }

    /**
     * 功能描述:
     * <消费者订阅消息>
     *
     * @param channel     1
     * @param declareOk   2
     * @param consumerTag 3
     * @return void
     * @author zhoulipu
     * @date 2019/7/31 17:51
     */
    public static void subscribe(Channel channel, AMQP.Queue.DeclareOk declareOk, String consumerTag) throws IOException {
        // 消费者订阅消息 监听队列 (队列名, 是否自动应答(与消息可靠有关 后续会介绍), 消费者标签, 消费者)
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                System.out.println("consumerTag:" + consumerTag);
                System.out.println("envelope:" + envelope.toString());
                System.out.println("properties:" + properties.toString());
                System.out.println("消息内容:" + new String(body));
            }
        };
        // 回调（队列名，是否自动确认消息，消费者）
        channel.basicConsume(declareOk.getQueue(), true, consumerTag, defaultConsumer);
    }

    /**
     * 功能描述:
     * <消费者配置>
     *
     * @param channel    1
     * @param pbxName    2
     * @param routingKey 3
     * @param queueName  4
     * @return com.rabbitmq.client.AMQP.Queue.DeclareOk
     * @author zhoulipu
     * @date 2019/7/31 17:51
     */
    public static AMQP.Queue.DeclareOk consumerConfig(Channel channel, String pbxName, String routingKey, String queueName) throws IOException {
        // 消息过期配置（文末对参数有简述）
        Map<String, Object> arguments = new HashMap<String, Object>();
        // 声明队列 (队列名, 是否持久化, 是否排他, 是否自动删除, 队列中的消息自动删除配置);
        AMQP.Queue.DeclareOk declareOk = channel.queueDeclare(queueName, true, false, false, arguments);
        // 将队列Binding到交换机上 (队列名, 交换机名, 路由关键字);
        channel.queueBind(declareOk.getQueue(), pbxName, routingKey);
        return declareOk;
    }
    
}