RabbitMQ消息中间件

时间  2019-11-30
标签 rabbitmq 消息 中间件 栏目 RabbitMQ 繁體版
原文   原文链接

消息中间件简单介绍java

消息中间件（消息队列）是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题实现高性能，高可用，可伸缩和最终一致性[架构] 使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ
 
那什么是RabbitMQ呢？
  • RabbitMQ是一个由Erlang语言发开的AMQP的开源实现
  • AMQP ：Advanced Message Queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。
  • RabbitMQ 最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。具体特色包括：
可靠性（Reliability） 使用了一些机制来保证可靠性，如持久、传输确认、发布确认
灵活性（Flexible Routing） 在消息进入队列以前，经过Exchange来路由消息，能够将多个Exchange绑定一块儿
消息集群（Clustering） 多个RabbitMQ服务器开以组成一个集群，造成一个逻辑Broker
高可用（Highly Available Queues） 队列能够在集群中机器上进行镜像，使得在部分节点出问题的状况下队列仍能够用
多中协议（Multi-protocol） 支持多种消息队列协议，好比STOMP、MQTT等等
多语言客户端（Many Clients） 几乎支持全部经常使用语言，好比java、.NET、Ruby等等
管理界面（Management UI） 提供了一个易用的用户界面，使得用户能够监控和管理消息Broker的许多方面
跟踪机制（Tracing） 若是消息异常，RabbitMQ提供了消息跟踪机制，使用者能够找出发生了什么
插件机制（Plugin System） 提供了许多插件，从多方面进行扩展，也可编写本身的插件

架构图redis

主要概念spring

RabbitMQ Server： 也叫broker server，它是一种传输服务。 他的角色就是维护一条从Producer到Consumer的路线，保证数据可以按照指定的方式进行传输。
Producer： 消息生产者，如图A、B、C，数据的发送方。消息生产者链接RabbitMQ服务器而后将消息投递到Exchange。
Consumer：消息消费者，如图一、二、3，数据的接收方。消息消费者订阅队列，RabbitMQ将Queue中的消息发送到消息消费者。
Exchange：生产者将消息发送到Exchange（交换器），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。Exchange并不存储消息。RabbitMQ中的Exchange有direct、fanout、topic、headers四种类型，每种类型对应不一样的路由规则。
Queue：（队列）是RabbitMQ的内部对象，用于存储消息。消息消费者就是经过订阅队列来获取消息的，RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者能够从Queue中获取消息并消费。多个消费者能够订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每一个消费者都收到全部的消息并处理。
RoutingKey：生产者在将消息发送给Exchange的时候，通常会指定一个routing key，来指定这个消息的路由规则，而这个routing key须要与Exchange Type及binding key联合使用才能最终生效。在Exchange Type与binding key固定的状况下（在正常使用时通常这些内容都是固定配置好的），咱们的生产者就能够在发送消息给Exchange时，经过指定routing key来决定消息流向哪里。RabbitMQ为routing key设定的 长度限制为255bytes。
Connection： （链接）：Producer和Consumer都是经过TCP链接到RabbitMQ Server的。之后咱们能够看到，程序的起始处就是创建这个TCP链接。
Channels： （信道）：它创建在上述的TCP链接中。数据流动都是在Channel中进行的。也就是说，通常状况是程序起始创建TCP链接，第二步就是创建这个Channel。
VirtualHost：权限控制的基本单位，一个VirtualHost里面有若干Exchange和MessageQueue，以及指定被哪些user使用 
 
RabbitMQ安装与启动
（1）选择相应的版本下载并安装 Eralng
（2）下载好后双击安装， 注意不要安装在包含中文和空格的目录下！安装后window服务中就存在rabbitMQ了，而且是启动状态。
（3）安装管理界面（插件）：进入rabbitMQ安装目录的sbin目录，输入命令
rabbitmq‐plugins enable rabbitmq_management 

（4）从新启动服务浏览器

（5）打开浏览器，地址栏输入http://127.0.0.1:15672 ,便可看到管理界面的登录页服务器

 
输入用户名和密码，都为guest 进入主界面：
 
RabbitMQ发送与接收消息
直接模式（Direct）
建立队列，名为queue.test
 
代码实现-消息生产者
（1）建立工程rabbitmq_demo，引入依赖 ，pom.xml以下： 
<dependency> 
     <groupId>org.springframework.amqp</groupId> 
     <artifactId>spring‐rabbit</artifactId>
     <version>2.1.4.RELEASE</version>
</dependency>
（2）编写配置文件applicationContext-rabbitmq-producer.xml  
<?xml version="1.0" encoding="UTF‐8"?> 
<beans xmlns="http://www.springframework.org/schema/beans"     
            xmlns:xsi="http://www.w3.org/2001/XMLSchema‐instance" 
            xmlns:rabbit="http://www.springframework.org/schema/rabbit" 
            xsi:schemaLocation="http://www.springframework.org/schema/beans 
            http://www.springframework.org/schema/beans/spring‐beans.xsd 
            http://www.springframework.org/schema/rabbit 
            http://www.springframework.org/schema/rabbit/spring‐rabbit.xsd"> 
    <!‐‐链接工厂‐‐> 
    <rabbit:connection‐factory id="connectionFactory"
　　　　　　　　host="127.0.0.1"
　　　　　　　　port="5672" 
　　　　　　　　username="guest" 
　　　　　　　　password="guest" />
    <rabbit:template id="rabbitTemplate" connection‐ factory="connectionFactory" /> 
</beans>
（3）消息生产者(发送消息) 
 1      @Test
　　　　　public void test() {
 2         //解析配置文件，从中获取RabbitTemplate对象
 3         ApplicationContext context=new ClassPathXmlApplicationContext("applicationContext-rabbitmq-producer.xml");
 4         RabbitTemplate rabbitTemplate= (RabbitTemplate)context.getBean("rabbitTemplate");
 5         //发送消息（queue.test这个队列名称须要在RabbitMQ中手动建立）
 6         rabbitTemplate.convertAndSend("","queue.test","直接模式");
 7         System.out.println("消息发送成功");
 8         //关闭对象
 9         ((ClassPathXmlApplicationContext) context).close();
10     }

代码实现、消息消费者架构

（1）编写消息监听类 
1 public class MessageConsumer implements MessageListener {
2     public void onMessage(Message message) {
3         System.out.println("接收到消息：" +new String(message.getBody()) );
4     }
5 }
（2）建立配置文件applicationContext-rabbitmq-consumer.xml 
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!--链接工厂-->
    <rabbit:connection-factory id="connectionFactory" host="127.0.0.1" port="5672" username="guest" password="guest"/>
    <!--队列名称-->
    <rabbit:queue name="queue.test" />
    <!--消费者监听类class为监听类路径-->
    <bean id="messageConsumer" class="cn.itcast.demo.MessageConsumer"> </bean>
    <!--设置监听容器-->
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queue-names="queue.test" ref="messageConsumer"/>
    </rabbit:listener-container>
</beans>
（3）编写测试代码Test2 
    @Test
    public void test2(){
        //运行此方法会自动调用消息监听类，并打印监听到的消息
        ApplicationContext context=new ClassPathXmlApplicationContext("applicationContext-rabbitmq-consumer.xml");
    }

运行结果app

接收到消息：直接模式

 

分列模式（Fanout）
1.这种模式须要提早将Exchange与Queue进行绑定，一个Exchange能够绑定多个Queue，一个Queue能够同多个Exchange进行绑定。
2.这种模式不须要RouteKey
3.若是接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。
代码实现（修改直接模式代码便可）
修改、消息生产者（发送消息） 
rabbitTemplate.convertAndSend("exchange.fanout_test","","分列模式走起");

修改消息、消费者配置文件（applicationContext-rabbitmq-consumer.xml）异步

    <!--队列名称-->
    <rabbit:queue name="queue.test" />
    <rabbit:queue name="queue.test1" />

    <!--消费者监听类class为监听类路径-->
    <bean id="messageConsumer" class="cn.itcast.demo.MessageConsumer"></bean>
    <bean id="messageConsumer1" class="cn.itcast.demo.MessageConsumer1"> </bean>

    <!--设置监听容器-->
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queue-names="queue.test" ref="messageConsumer"/>
        <rabbit:listener queue-names="queue.test1" ref="messageConsumer1"/>
    </rabbit:listener-container>

在增长一个消息监听者类，而后运行测试代码Test2便可分布式

运行结果性能

我是Message、接收到消息：分列模式走起

我是Message一、接收到消息：分列模式走起

 

死信交换器 Dead Letter Exchanges
一个消息在知足以下条件下，会进死信路由，记住这里是路由而不是队列，一个路由能够对应不少队列。
（1） 一个消息被Consumer拒收了，而且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其余消费者使用。
（2）消息存活时间到了，消息过时了。
（3）队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。
代码实现
建立配置文件applicationContext-mq.xml（ 交换器和队列是自动配置，无需手动配置） 
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!--配置connection-factory，指定链接rabbit server参数 -->
    <rabbit:connection-factory id="connectionFactory" username="guest" password="guest" host="127.0.0.1" port="5672" publisher-confirms="true"/>
    <rabbit:admin connection-factory="connectionFactory"></rabbit:admin>

    <!--建立交换机(过时队列的交换机)-->
    <rabbit:direct-exchange id="exchange.delay.order.begin" name="exchange.delay.order.begin" durable="false" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="queue.delay.order.begin" key="delay"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!-- 延时队列(过时队列) -->
    <rabbit:queue name="queue.delay.order.begin" durable="false">
        <rabbit:queue-arguments>
            <!-- 设置队列过时时间为1分钟 -->
            <entry key="x-message-ttl" value="60000" value-type="java.lang.Long"/>
            <entry key="x-dead-letter-exchange" value="exchange.delay.order.done"/>
            <entry key="x-dead-letter-routing-key" value="delay"/>
        </rabbit:queue-arguments>
    </rabbit:queue>

    <!--死信交换机定义-->
    <rabbit:direct-exchange id="exchange.delay.order.done" name="exchange.delay.order.done" durable="false" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="queue.delay.order.done" key="delay"/>
            <!-- binding key 相同为 【delay】exchange转发消息到多个队列 -->
            <!--<rabbit:binding queue="queue.delay.order.done.two" key="delay" />-->
        </rabbit:bindings>
    </rabbit:direct-exchange>
    <rabbit:queue name="queue.delay.order.done" durable="false"/>
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>

    <!-- 消息接收者 -->
    <rabbit:listener-container connection-factory="connectionFactory" channel-transacted="false">
        <rabbit:listener queues="queue.delay.order.done" ref="orderMessageListener"/>
    </rabbit:listener-container>

    <!--建立一个消息监听对象-->
    <bean class="cn.itcast.demo.MessageConsumer" id="orderMessageListener"></bean>
</beans>

建立测试类消息生产者（发送消息）

    @Test
    public void test1() {
        //解析配置文件，从中获取RabbitTemplate对象
        ApplicationContext context=new ClassPathXmlApplicationContext("applicationContext-mq.xml");
        RabbitTemplate rabbitTemplate= (RabbitTemplate)context.getBean("rabbitTemplate");
        rabbitTemplate.convertAndSend("exchange.delay.order.begin","delay","延时队列模式走起");
        System.out.println("消息发送成功");
        //关闭对象
        ((ClassPathXmlApplicationContext) context).close();
    }

建立消息监听器（MessageConsumer）

public class MessageConsumer implements MessageListener {
    public void onMessage(Message message) {
        System.out.println("我是Message、接收到消息：" +new String(message.getBody()) );
    }
}

建立测试类消息消费者

    @Test
    public void test2(){
        //运行此方法会自动调用消息监听类，并打印监听到的消息
        ApplicationContext context=new ClassPathXmlApplicationContext("applicationContext-mq.xml");
    }

运行结果（因为死信队列时间设置为1分钟，因此须要一分钟后才会监听到消息）

我是Message、接收到消息：延时队列模式走起

 

如何解决消息重复消费

让生产者发送每条数据的时候，里面加一个全局惟一的id，相似订单id之类的东西，而后你这里消费到了以后，先根据这个id去好比redis里查一下，以前消费过吗？若是没有消费过，你就处理，而后这个id写redis。若是消费过了，那你就别处理了，保证别重复处理相同的消息便可。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程