java-web系列（八）---RabbitMQ在java-web中的简单应用

墨蓝 2022-05-08 05:34 248阅读 0赞

# 前言 #

这个项目的github地址：[extensible项目的github地址][extensible_github]

extensible项目当前功能模块如下：

[java-web系列（一）—搭建一个基于SSM框架的java-web项目][java-web_SSM_java-web]  
[java-web系列（二）—以dockerfile的方式发布java-web项目][java-web_dockerfile_java-web]  
[java-web系列（三）—(slf4j + logback)进行日志分层][java-web_slf4j _ logback]  
[java-web系列（四）—几种常见的加密算法][java-web]  
[java-web系列（五）—SpringBoot整合Redis][java-web_SpringBoot_Redis]  
[java-web系列（六）—Mybatis动态多数据源配置][java-web_Mybatis]  
[java-web系列（七）—SpringBoot整合Quartz实现多定时任务][java-web_SpringBoot_Quartz]  
[java-web系列（八）—RabbitMQ在java-web中的简单应用][java-web_RabbitMQ_java-web]  
[java-web系列（九）—SpringBoot整合ElasticSearch][java-web_SpringBoot_ElasticSearch]

如对该项目有疑问，可在我的博客/github下面留言，也可以以邮件的方式告知。  
我的联系方式：dzy930724@163.com

# RabbitMQ的使用场景 #

MQ，是Message Queue(消息队列)的简写。简而言之，RabbitMQ就是将消息储存在队列中。  
  
在项目的实际开发过程中，可以将一些无需即时返回结果且耗时的操作提取出来，进行异步处理。这种处理方式能够大大节省服务器的请求响应时间，从而提高系统的吞吐量。

比如：以去年双十一淘宝成交额为例

![2017年双十一成交额][2017]

**当天每秒下订单笔数超过32.5万笔，支付笔数超过25.6万笔。**  
  
也就是说，需要阿里的服务器每秒进行32.5万个“生成订单”的操作，还要进行25.6万个“支付订单”的操作。  
  
而这些与money相关的，都是一些耗时的操作。如果要求即时返回这些操作的处理结果，服务器的压力太大。  
  
实际能够进行优化后的做法是：把两种不同的操作，放到两个不同的队列中延后进行处理。  
  
如：“剁手党”之一的我在当天进行了一个下单操作，服务器会将这个操作的具体逻辑放到“order\_queue”队列中，就告知我下单成功，然后服务器会在空闲时处理并生成相应的订单。这种做法，不仅提升了用户的使用体验（我觉得很快就下单成功了），还能够缓解服务器的压力。

常用的消息队列还有：ZeroMQ,ActiveMQ,Kafka等等，有兴趣的可以自行了解。这篇博客主要是学习RabbitMQ的简单应用，这里就不探讨这几种消息队列的优劣势了。

# 在Windows下安装RabbitMQ #

> RabbitMQ是使用Erlang语言编写的一个开源的消息队列。

下面的这两个安装包也可以在[我的网盘][Link 1]中下载,提取码为：bs7t。

1.  首先，在[Erlang官网][Erlang]下载对应版本的Erlang平台。  
    ![erlang依赖][erlang]
2.  然后运行可执行文件(otp\_win64\_21.1)。按默认配置进行安装，更改一下文件的存放目录即可。出现如下图说明Erlang平台安装成功。  
    ![erlang平台安装成功][erlang 1]
3.  再然后在[RabbitMQ官网][RabbitMQ]下载最新版本的RabbitMQ Server。  
    ![RabbitMQ Server下载][RabbitMQ Server]
4.  继续按默认配置进行安装，出现如下图说明RabbitMQ Server安装成功。  
    ![RabbitMQ Server安装成功][RabbitMQ Server 1]
5.  这种做法，是将RabbitMQ暴露为Window的一个服务。在这里启动RabbitMQ后，可以在15672端口用默认账户密码进行登录（guest/guest）。  
    ![默认账户guest登录RabbitMQ Server][guest_RabbitMQ Server]
6.  出现如下图说明登录成功。  
    ![RabbitMQ Server主页][RabbitMQ Server 2]

# RabbitMQ中的几个重要概念 #

*  Producer

Producer，生产者。消息的发送方即为生产者。

*  Consumer

Consumer，消费者。消息的接收方即为消费者。

*  Connection

Connection，获取RabbitMQ(消息队列)服务的长连接。得到长连接实例的方式有两种:

1.  手动设置相关参数（host/port etc…）

/** * 不设置的话，会使用默认的参数（userName:guest,password:guest,virtualHost:/,hostName:localhost,portNumber:5672） */
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUsername(userName);
    factory.setPassword(password);
    factory.setVirtualHost(virtualHost);
    factory.setHost(hostName);
    factory.setPort(portNumber);
    Connection conn = factory.newConnection();

1.  URI设置相关参数

/** * 不设置的话，会使用默认的参数（userName:guest,password:guest,virtualHost:/,hostName:localhost,portNumber:5672） */
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost");
    Connection connection = factory.newConnection();

*  Channel

Channel，通道。通过通道来决定“生产者如何往队列中发送消息”、“消费者如何从队列中接收消息”。获取通道实例的方式如下：

Channel channel = connection.createChannel();

*  Exchange

Exchange，交换机。生产者往队列中发送消息，实际不是直接发送给队列，而是先发送给交换机，由交换机决定实际将消息发送给哪个队列。

*  Queue

Queue，队列。存储消息的地方。交换机和队列都是通过通道实例来声明，而且交换机与队列直接是存在对应关系的。具体的源码如下：

channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true);
    channel.queueDeclare(queueName,true,false,false,null);
    channel.queueBind(queueName, exchangeName, routingKey);

具体的代码解释如下：

通过`ChannelIN.exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable)`来声明交换机实例，第一个参数是交换机的名称，第二个参数是交换机的类型（direct/fanout/topic/headers）,第三个参数是该交换机是否持久化。  
  
通过`ChannelIN.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)`来声明队列实例，第一个参数是队列的名称，第二个参数是该队列是否持久化，第三个参数是是否只有自己能够看到该对列（排他性），第四个参数是当没有消费者占用该队列时是否删除该队列。  
  
通过`queueBind(String queue, String exchange, String routingKey)`绑定该交换机和队列。routingKey是绑定队列和交换机之间的路由规则。

这些概念之间的关系结构图如下：

![RabbitMQ的元素结构图][RabbitMQ 1]

# RabbitMQ在java-web中的应用 #

*  必须要导入RabbitMQ Server的POM依赖

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

*  发送消息的生产者测试代码如下：

/** * RabbitMQ中生产者测试代码 * @author zhenye 2018/9/29 */
    @Slf4j
    public class ProducerTest { 
    
        private final static String EXCHANGE_NAME = "MY_EXCHANGE";
        private final static String QUEUE_NAME = "MY_QUEUE";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT,true);
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            String routingKey = "123";
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,routingKey);
            String message = "Hello RabbitMQ, I will send some message to the consumer.";
            channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));
            log.info("Producer send message, the content is :" + message);
            channel.close();
            connection.close();
        }
    }

*  接收消息的消费者测试代码如下：

/** * RabbitMQ中消费者测试代码 * @author zhenye 2018/9/29 */
    @Slf4j
    public class ConsumerTest { 
        private final static String QUEUE_NAME = "MY_QUEUE";
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            log.info("Consumer is waiting!");
            Consumer consumer = new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String message = new String(body,"UTF-8");
                    log.info("Consumer received message, the content is:" + message);
                }
            };
            channel.basicConsume(QUEUE_NAME,true,consumer);
        }
    }

*  演示过程

连续运行5次`ProducerTest.main()`,即有5个消费者往队列（MY\_QUEUE）中发送了消息。

RabbitMQ Server主页显示如下：

![主页效果图1][1]

结果表明，在RabbitMQ的所有队列中,还有5条待处理的消息。

接着运行`ConsumerTest.main()`，效果图如下：

![主页效果图2][2]

结果表明，该消费者一次性处理完了5条消息。

*  流程说明

生产者是通过`ChannelIN.basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)`来发送消息，第一个参数是交换机名称，第二个参数是路由规则Key,第三个参数是全局属性，第四个参数是消息实体。第一、二个参数是可以确定要保存消息的队列的。

消费者是通过`DefaultConsumer.handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)`来接收并处理消息的。由于消费者是直接从队列中获取消息，只要保证生产者存入消息的队列（交换机名称、路由Key确定的队列），与消费者接收消息的队列（直接指定QUEUE\_NAME）相同，就能正确地取出消息。

[extensible_github]: https://github.com/zhenye163/extensible
[java-web_SSM_java-web]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82150584
[java-web_dockerfile_java-web]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82150811
[java-web_slf4j _ logback]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82151006
[java-web]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82429467
[java-web_SpringBoot_Redis]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82594054
[java-web_Mybatis]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82787572
[java-web_SpringBoot_Quartz]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82867193
[java-web_RabbitMQ_java-web]: https://blog.csdn.net/UtopiaOfArtoria/article/details/82981979
[java-web_SpringBoot_ElasticSearch]: https://blog.csdn.net/UtopiaOfArtoria/article/details/88403459
[2017]: /images/20220508/7e20ba98cb514a98be658f7f5905db25.png
[Link 1]: https://pan.baidu.com/s/1tDKh21XR1uY4HDm4TU4OPA
[Erlang]: http://www.erlang.org/download.html
[erlang]: /images/20220508/c95e811eff324fddb5f8570a5ba1c316.png
[erlang 1]: /images/20220508/708ba12642654cceb2abcd0545c46d7f.png
[RabbitMQ]: http://www.rabbitmq.com/install-windows.html
[RabbitMQ Server]: /images/20220508/d942ac8094bf4a11b64de0a821ca577a.png
[RabbitMQ Server 1]: /images/20220508/9fb8178fb3154d1fa54124da89c0d567.png
[guest_RabbitMQ Server]: /images/20220508/829f26b694d04aa98c822b92d9b35fe5.png
[RabbitMQ Server 2]: /images/20220508/529f34e92c0a474bb8aff60454fb38dc.png
[RabbitMQ 1]: /images/20220508/c54b2ccc0d544d53a1fc5e95ce0c619d.png
[1]: /images/20220508/ee68f0e3e4874d93beb4ba1616305120.png
[2]: /images/20220508/a1f07c81fedb4351934387d96ac1bee0.png