全网首发，做第一人纯源码讲解 RabbitMQ 实践，收藏吧

关注
发布于: 2020 年 10 月 31 日
前言之前一直用kafka，今天不是正好周五啊，我本来又可以摸鱼了，老大说，这周别摸鱼了，我还想反驳一下，老大说：行了，你小子别逼逼，你就是在摸鱼，今天你把mq的东西稍微地整理一下，下个星期的交流会你来主持，我要出差，你小子要给我出岔子，能死你，没办法，只能苦兮兮地去研究，正好也不知道周末该写什么，我就把RabbitMQ的简单应用、工作队列模式、发布/订阅模式、路由模式、通配符模式五种模式附源码罗列出来，希望对正在学习的大家有所帮助
注：因为篇幅原因，路由模式和通配符模式这里展现不开，我留到明天在展示，关注我，一个神奇的程序猿
我已经不行了，下班了，找地方喝两杯去，原创不易，一下午没干别的啊，整理出来，希望大家关注+转发一下，谢谢，给我一点支持，有问题的可以下方评论区联系我
个人公众号：Java架构师联盟，每日更新技术好文
好了，下班，大家看正经的吧
RabbitMQ（一）：Java简单操作RabbitMQ一、添加pom依赖
<dependency>
   <groupId>com.rabbitmq</groupId>
   <artifactId>amqp-client</artifactId>
</dependency>
1234
二、生产者
public class Send {
    // 定义一个消息队列名称
    private final static String QUEUE_NAME = "FIRST-MQ";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 建立新的连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq
        factory.setVirtualHost("/");
        factory.setUsername("user");
        factory.setPassword("pwd");
        Connection connection = null;
        Channel channel = null;
        try{
            connection = factory.newConnection();
            // 创建内部会话通道，生产者和mq服务通信都在channel中完成
            channel = connection.createChannel();
            // 声明交换机，这里暂时没有
﻿
            // 声明创建队列，如果没有，则创建
            // 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            /**
             * queue：队列名称
             * durable：是否持久化，存到磁盘中
             * exclusive:是否排他，是否独占连接，该队列只允许在这个connection中访问，如果关闭连接，则删除队列
             * autoDelete:不再使用时自动删除。和exclusive 一同设为true 可以实现临时队列。
             * args：扩展参数，比如存活时间
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false, null);
            String message = "HELLO WORLD";
            // 发送消息
            // 参数：String exchange, String routingKey, BasicProperties props, byte[] body
            /**
             * exchange:交换机 如果不指定("")，就默认交换机
             * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称
             * props:额外属性
             * body：消息内容
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("send message:"+message);
        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
        } finally {
            // 先关闭通道
            channel.close();
            // 再关闭连接
            connection.close();
        }
    }
}
​
三、消费者
public class Recv {
​
    private final static String QUEUE_NAME = "FIRST-MQ";
​
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.*5.**.8*");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("user");
        factory.setPassword("pwd");
​
        try{
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" Waiting for messages. ");
​
            // 定义实现消费方法
            Consumer consumer = new DefaultConsumer(channel) {
                /**
                 * 当接收到消息后，此方法将被调用
                 * @description:
                 * @param consumerTag: (非必须)消费者标签：标识消费者，在监听队列的时候设置channel.basicConsume中设置
                 * @param envelope: 信封，通过envelope 可以获取很多信息，交换机、消息id(可用于消息接收ack)、路由Key等等
                 * @param properties:消息的属性，发送消息时 设置的额外属性
                 * @param body :消息内容
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, StandardCharsets.UTF_8);
                    System.out.println("receive:" + message);
                    TimeUnit.SECONDS.sleep(1);
                    // 消息id
                    envelope.getDeliveryTag();
                    // 手动回复ack
                    // channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_NAME, true, consumer);
            TimeUnit.SECONDS.sleep(15);
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
﻿
工作队列模式(work queue)原理
一个消息轮流被接收
﻿
特点：
一个生产者发送消息
多个消费者共同监听一个队列的消息
消息不能被重复消费
采用轮询的方式将消息平均发送给消费者(c1,c2,c1,c2…交替)
引入pom
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>
1234
生产者
public class Send {
    // 队列名
    private final static String QUEUE_NAME = "FIRST-MQ";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 建立新的连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = null;
        Channel channel = null;
        try{
            connection = factory.newConnection();
            // 创建内部会话通道，生产者和mq服务通信都在channel中完成
            channel = connection.createChannel();
            // 声明交换机，这里暂时没有,就会用默认的交换机
​
            // 声明创建队列，如果没有，则创建
            // 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            /**
             * queue：队列名称
             * durable：是否持久化，存到磁盘中
             * exclusive:是否排他，是否独占连接，该队列只允许在这个connection中访问，如果关闭连接，则删除队列
             * autoDelete:不再使用时自动删除。和exclusive 一同设为true 可以实现临时队列。
             * args：扩展参数，比如存活时间
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false, null);
            String message = "work queue 模式，消息来了。";
            // 发送消息
            // 参数：String exchange, String routingKey, BasicProperties props, byte[] body
            /**
             * exchange:交换机 如果不指定("")，就默认交换机
             * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称
             * props:额外属性
             * body：消息内容
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("send message:"+message);
        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
        } finally {
            // 先关闭通道
            channel.close();
            // 再关闭连接
            connection.close();
        }
    }
}
​
消费者1
public class Recv {
    private final static String QUEUE_NAME = "FIRST-MQ";
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
​
        try{
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" [*] Waiting for messages.");
​
            // 定义实现消费方法
            Consumer consumer = new DefaultConsumer(channel) {
                /**
                 * 当接收到消息后，此方法将被调用
                 * @description:
                 * @param consumerTag: (非必须)消费者标签：标识消费者，在监听队列的时候设置channel.basicConsume中设置
                 * @param envelope: 信封，通过envelope 可以获取很多信息，交换机、消息id(可用于消息接收ack)、路由Key等等
                 * @param properties:消息的属性，发送消息时 设置的额外属性
                 * @param body :消息内容
                 */
                @SneakyThrows
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, StandardCharsets.UTF_8);
                    System.out.println("消费者1 receive:" + message);
                    TimeUnit.SECONDS.sleep(1);
                    // 手动回复ack
                    // channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_NAME, true, consumer);
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
​
消费者2
public class Recv2 {
    private final static String QUEUE_NAME = "FIRST-MQ";
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
​
        try{
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" [*] Waiting for messages.");
​
            // 定义实现消费方法
            Consumer consumer = new DefaultConsumer(channel) {
                /**
                 * 当接收到消息后，此方法将被调用
                 * @description:
                 * @param consumerTag: (非必须)消费者标签：标识消费者，在监听队列的时候设置channel.basicConsume中设置
                 * @param envelope: 信封，通过envelope 可以获取很多信息，交换机、消息id(可用于消息接收ack)、路由Key等等
                 * @param properties:消息的属性，发送消息时 设置的额外属性
                 * @param body :消息内容
                 */
                @SneakyThrows
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, StandardCharsets.UTF_8);
                    System.out.println("消费者2 receive:" + message);
                    TimeUnit.SECONDS.sleep(1);
                    // 手动回复ack
                    // channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_NAME, true, consumer);
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
​
先启动两个消费者，然后再多次启动生产者，就能看到效果啦。
发布/订阅模式(Publish/Subscribe)一个消息 ， 多个接收
﻿
特点：
一个生产者将消息发送给交换机
与交换机绑定的有多个队列，每个消费者监听自己的队列
生产者将消息发送给交换机，交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。
如果消息发送给没有绑定队列的交换机上，消息将丢失。
他比工作队列模式(work queue)更为强大，可以看成多个工作队列。
引入pom
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
</dependency>
1234
生产者
// 这里模拟注册完成后，发送短信和邮件。
public class Producer {
    // 定义两个队列和一个交换机
    private final static String QUEUE_EMAIL = "queue_email";
    private final static String QUEUE_SMS = "queue_sms";
    private final static String EXCHANGE_NAME = "exchange_fanout_1";
​
    public static void main(String[] args) throws IOException, TimeoutException {
​
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = null;
        Channel channel = null;
        try{
            connection = factory.newConnection();
            // 创建内部会话通道，生产者和mq服务通信都在channel中完成
            channel = connection.createChannel();
            // 声明交换机
            /**
             * 1. 交换机名称
             * 2、交换机类型:
             *  FANOUT:对应的模式就是 发布/订阅模式
             *  DIRECT:对应 路由(Routing) 的工作模式
             *  TOPIC:对应 Topics 工作模式
             *  HEADERS: 对应 HEADERS 工作模式
             */
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 声明创建队列，如果没有，则创建
            channel.queueDeclare(QUEUE_EMAIL,false,false,false, null);
            channel.queueDeclare(QUEUE_SMS,false,false,false,null);
​
            // 交换机队列绑定
            /**
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key 发布订阅模式中 设置为""
             */
            channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,"");
            channel.queueBind(QUEUE_SMS,EXCHANGE_NAME,"");
​
            String message = "接好了，SMS和EMAIL~~";
            // 发送消息
            // 参数：String exchange, String routingKey, BasicProperties props, byte[] body
            /**
             * exchange:交换机 如果不指定("")，就默认交换机
             * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称
             * props:额外属性
             * body：消息内容
             */
            channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
            System.out.println("send message: "+message);
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            channel.close();
            connection.close();
        }
    }
}
​
消费者·接收短信
/**
 * 短信 消费者
 */
public class Consumer_SMS {
​
    private final static String QUEUE_EMAIL = "queue_email";
    private final static String QUEUE_SMS = "queue_sms";
    private final static String EXCHANGE_NAME = "exchange_fanout_1";
​
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
​
        Connection connection = null;
        Channel channel = null;
        try{
            connection = factory.newConnection();
            // 创建内部会话通道，生产者和mq服务通信都在channel中完成
            channel = connection.createChannel();
            // 声明交换机
            /**
             * 1. 交换机名称
             * 2、交换机类型:
             *  FANOUT:对应的模式就是 发布/订阅模式
             *  DIRECT:对应路由的工作模式
             *  TOPIC:对应 Topics 工作模式
             *  HEADERS: 对应 HEADERS 工作模式
             */
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 声明队列，如果没有，则创建
//            channel.queueDeclare(QUEUE_EMAIL,false,false,false, null);
            channel.queueDeclare(QUEUE_SMS,false,false,false,null);
​
            // 交换机队列绑定
            /**
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key 发布订阅模式中 设置为""
             */
//            channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,"");
            channel.queueBind(QUEUE_SMS,EXCHANGE_NAME,"");
​
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String (body, StandardCharsets.UTF_8);
                    System.out.println("SMS receive:" + message);
                }
            };
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_SMS,true,consumer);
        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
        }
    }
}
消费者·接收邮件
public class Consumer_EMAIL {
    private final static String QUEUE_EMAIL = "queue_email";
    private final static String QUEUE_SMS = "queue_sms";
    private final static String EXCHANGE_NAME = "exchange_fanout_1";
​
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
​
        Connection connection = null;
        Channel channel = null;
​
        try{
            connection = factory.newConnection();
            // 创建内部会话通道，生产者和mq服务通信都在channel中完成
            channel = connection.createChannel();
            // 声明交换机
            /**
             * 1. 交换机名称
             * 2、交换机类型:
             *  FANOUT:对应的模式就是 发布/订阅模式
             *  DIRECT:对应路由的工作模式
             *  TOPIC:对应 Topics 工作模式
             *  HEADERS: 对应 HEADERS 工作模式
             */
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 声明队列，如果没有，则创建
            channel.queueDeclare(QUEUE_EMAIL,false,false,false, null);
​
            // 交换机队列绑定
            /**
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key 发布订阅模式中 设置为""
             */
            channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,"");
​
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String (body, StandardCharsets.UTF_8);
                    System.out.println("Email receive:" + message);
                }
            };
​
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_EMAIL,true,consumer);
        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
        }
    }
}
​
补充：结合工作队列模式其实只要在 消费者【c1】并列加上一个【c3】即可，c1、c3共享同一队列
// 新增消费者 Consumer_SMS_2
public class Consumer_SMS_2 {
    private final static String QUEUE_EMAIL = "queue_email";
    private final static String QUEUE_SMS = "queue_sms";
    private final static String EXCHANGE_NAME = "exchange_fanout_1";
​
    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("106.**.**.82");
        factory.setPort(5672);
        // 设置虚拟主机 一个mq的服务可以设置多个虚拟机，每个虚拟机相当于独立的mq
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
​
        Connection connection = null;
        Channel channel = null;
        try{
            connection = factory.newConnection();
            // 创建内部会话通道，生产者和mq服务通信都在channel中完成
            channel = connection.createChannel();
            // 声明交换机
            /**
             * 1. 交换机名称
             * 2、交换机类型:
             *  FANOUT:对应的模式就是 发布/订阅模式
             *  DIRECT:对应路由的工作模式
             *  TOPIC:对应 Topics 工作模式
             *  HEADERS: 对应 HEADERS 工作模式
             */
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            // 声明队列，如果没有，则创建
//            channel.queueDeclare(QUEUE_EMAIL,false,false,false, null);
            channel.queueDeclare(QUEUE_SMS,false,false,false,null);
​
            // 交换机队列绑定
            /**
             * 1 queue 队列名称
             * 2 exchange 交换机名称
             * 3 routingKey 路由Key 发布订阅模式中 设置为""
             */
//            channel.queueBind(QUEUE_EMAIL,EXCHANGE_NAME,"");
            channel.queueBind(QUEUE_SMS,EXCHANGE_NAME,"");
​
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String (body, StandardCharsets.UTF_8);
                    System.out.println("SMS_2 receive:" + message);
                }
            };
            // 接收消息 监听队列
            // 参数：String queue, boolean autoAck, Consumer callback
            /**
             * queue:队列
             * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复
             * callback:消费方法，当消费者接收消息执行的方法。
             */
            channel.basicConsume(QUEUE_SMS,true,consumer);
        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
        }
    }
}
﻿
发布于: 2020 年 10 月 31 日阅读数: 25
原文链接:【http://xie.infoq.cn/article/2f963ff52753da85962ee5f35】。
小Q

关注
还未添加个人签名 2020.06.30 加入
小Q 公众号：Java架构师联盟作者多年从事一线互联网Java开发的学习历程技术汇总，旨在为大家提供一个清晰详细的学习教程，侧重点更倾向编写Java核心内容。如果能为您提供帮助，请给予支持(关注、点赞、分享)！
发布
暂无评论
创作场景

全网首发，做第一人纯源码讲解 RabbitMQ 实践，收藏吧

前言

RabbitMQ（一）：Java简单操作RabbitMQ

工作队列模式(work queue)

发布/订阅模式(Publish/Subscribe)

补充：结合工作队列模式

小Q

评论
