RabbitMQ 学习-direct模式

2018-12-17  本文已影响0人  松松木tell

RabbitMQ direct模式

关于它的快速入门，可以在这里查看：http://www.rabbitmq.com/getstarted.html。

生产者

1.配置文件

<?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-3.1.xsd
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.3.xsd">
    
    <!-- 连接服务配置 -->
    <rabbit:connection-factory id="rabbitConnectionFactory" addresses="${base-mq.addresses}"
                               virtual-host="${base-mq.vhost}" username="${base-mq.username}"
                               password="${base-mq.password}"/>
    <rabbit:admin connection-factory="rabbitConnectionFactory"/>
    <!-- template申明 -->
    <rabbit:template id="amqpTemplate" exchange="myTest-exchangeFirst"  connection-factory="rabbitConnectionFactory"/>

    <!-- 队列 -->
    <rabbit:queue id="testQueuea" name="testQueuea" durable="true"  auto-declare="true" auto-delete="false"/>
    <rabbit:queue id="testQueueb" name="testQueueb" durable="true"  auto-declare="true" auto-delete="false"/>

    <!-- 申明exchange -->
    <rabbit:direct-exchange id="myTest-exchangeFirst" name="myTest-exchangeFirst" durable="true" auto-declare="true" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="testQueuea" key="keya"/>
            <rabbit:binding queue="testQueueb" key="keyb"/>
            <rabbit:binding queue="testQueueb" key="keyab"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>
</beans>
  1. 生产者代码
package com.zjs.sp.rabbit.web.controller;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

@Controller
@RequestMapping(value = "/attribute")
public class RabbitProducerController {
    @Autowired
    private AmqpTemplate amqpTemplate;

    @RequestMapping("/hello")
    public String hello() {
        
        amqpTemplate.convertAndSend("keya", "消息a1");
        amqpTemplate.convertAndSend("keya", "消息a2");
        amqpTemplate.convertAndSend("keya", "消息a3");
        
        amqpTemplate.convertAndSend("keyb", "消息b1");
        amqpTemplate.convertAndSend("keyb", "消息b2");
        amqpTemplate.convertAndSend("keyb", "消息b3");
        
        amqpTemplate.convertAndSend("keyab", "消息ab");
        System.out.println("发送消息成功");
        return "Hello";
    }
}

消费者

  1. 配置文件
    applicationContext-springmq.xml
<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-3.1.xsd
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.3.xsd">
    
    <!-- 连接配置 -->
    <rabbit:connection-factory id="rabbitConnectionFactory" addresses="${base-mq.addresses}"
                               virtual-host="${base-mq.vhost}" username="${base-mq.username}" password="${base-mq.password}"/>
    <rabbit:admin connection-factory="rabbitConnectionFactory"/>
    
    <!-- 队列 -->
    <rabbit:queue id="testQueuea" name="testQueuea" durable="true"  auto-declare="true" auto-delete="false"/>
    <rabbit:queue id="testQueueb" name="testQueueb" durable="true"  auto-declare="true" auto-delete="false"/>
    <!-- 监听类 -->
    <bean id="consumeraListener" class="com.zjs.sp.consumer.web.listener.ConsumeraListener"/>
    <bean id="consumerbListener" class="com.zjs.sp.consumer.web.listener.ConsumerbListener"/>
    <!--     消费queue的listener -->
    <rabbit:listener-container connection-factory="rabbitConnectionFactory" acknowledge="auto" concurrency="1" >
        <rabbit:listener queues="testQueuea" method="onMessage" ref="consumeraListener"/>
         <rabbit:listener queues="testQueueb" method="onMessage" ref="consumerbListener"/>
    </rabbit:listener-container>

</beans>
  1. Listener代码

ConsumeraListener .java

package com.zjs.sp.consumer.web.listener;  

/**
 * @author zhangjinsong
 */
public class ConsumeraListener {
    public void onMessage(String message)  {
        System.out.println("AListener接收到消息："+message);
    }
}

ConsumerbListener .java

package com.zjs.sp.consumer.web.listener;  

/**
 * @author zhangjinsong
 */
public class ConsumerbListener {
    public void onMessage(String message)  {
        System.out.println("BListener接收到消息："+message);
    }
}

执行结果

发送消息成功
AListener接收到消息：消息a1
BListener接收到消息：消息b1
AListener接收到消息：消息a2
BListener接收到消息：消息b2
AListener接收到消息：消息a3
BListener接收到消息：消息b3
BListener接收到消息：消息ab

RabbitMQ Management绑定关系如下

image.png

总结

1.一般在direct模式中生产端申明绑定，由消费端只负责监听

        amqpTemplate.convertAndSend("keya", "消息a1");
        amqpTemplate.convertAndSend("keya", "消息a2");
        amqpTemplate.convertAndSend("keya", "消息a3");
        
        amqpTemplate.convertAndSend("keyb", "消息b1");
        amqpTemplate.convertAndSend("keyb", "消息b2");
        amqpTemplate.convertAndSend("keyb", "消息b3");
        
        amqpTemplate.convertAndSend("keyab", "消息ab");

生产端配置：

    <!-- 申明exchange -->
    <rabbit:direct-exchange id="myTest-exchangeFirst" name="myTest-exchangeFirst" durable="true" auto-declare="true" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="testQueuea" key="keya"/>
            <rabbit:binding queue="testQueueb" key="keyb"/>
            <rabbit:binding queue="testQueueb" key="keyab"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

消费端配置：

    <!-- 队列 -->
    <rabbit:queue id="testQueuea" name="testQueuea" durable="true"  auto-declare="true" auto-delete="false"/>
    <rabbit:queue id="testQueueb" name="testQueueb" durable="true"  auto-declare="true" auto-delete="false"/>
    <rabbit:queue id="testQueuec" name="testQueuec" durable="true"  auto-declare="true" auto-delete="false"/>
    <!-- 监听类 -->
    <bean id="consumeraListener" class="com.zjs.sp.consumer.web.listener.ConsumeraListener"/>
    <bean id="consumerbListener" class="com.zjs.sp.consumer.web.listener.ConsumerbListener"/>
    <bean id="consumercListener" class="com.zjs.sp.consumer.web.listener.ConsumercListener"/>
    <!--     消费queue的listener -->
    <rabbit:listener-container connection-factory="rabbitConnectionFactory" acknowledge="auto" concurrency="1" >
         <rabbit:listener queues="testQueuea" method="onMessage" ref="consumeraListener"/>
         <rabbit:listener queues="testQueueb" method="onMessage" ref="consumerbListener"/>
         <rabbit:listener queues="testQueuec" method="onMessage" ref="consumercListener"/>
    </rabbit:listener-container>

2.不管是生产端还是消费端，申明的exchange或者queue，如果名称相同的对象都是在rabbit服务器的同一个实例。
当然消费端同样可以再申明exchange ，如果名称一致不会重新创建，如果有新的绑定可以更新上去，例如消费端：

    <!-- 队列 -->
    <rabbit:queue id="testQueuea" name="testQueuea" durable="true"  auto-declare="true" auto-delete="false"/>
    <rabbit:queue id="testQueueb" name="testQueueb" durable="true"  auto-declare="true" auto-delete="false"/>
    <rabbit:queue id="testQueuec" name="testQueuec" durable="true"  auto-declare="true" auto-delete="false"/>
    <!-- 监听类 -->
    <bean id="consumeraListener" class="com.zjs.sp.consumer.web.listener.ConsumeraListener"/>
    <bean id="consumerbListener" class="com.zjs.sp.consumer.web.listener.ConsumerbListener"/>
    <bean id="consumercListener" class="com.zjs.sp.consumer.web.listener.ConsumercListener"/>
    <!--     消费queue的listener -->
    <rabbit:listener-container connection-factory="rabbitConnectionFactory" acknowledge="auto" concurrency="1" >
         <rabbit:listener queues="testQueuea" method="onMessage" ref="consumeraListener"/>
         <rabbit:listener queues="testQueueb" method="onMessage" ref="consumerbListener"/>
         <rabbit:listener queues="testQueuec" method="onMessage" ref="consumercListener"/>
    </rabbit:listener-container>
   <!--绑定 -->
    <rabbit:direct-exchange id="myTest-exchangeFirst" name="myTest-exchangeFirst" durable="true" auto-declare="true" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="testQueuec" key="keyab"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

RabbitMQ Management绑定关系如下


image.png

执行结果如下

发送消息成功
BListener接收到消息：消息b1
CListener接收到消息：消息ab
AListener接收到消息：消息a1
BListener接收到消息：消息b2
AListener接收到消息：消息a2
BListener接收到消息：消息b3
AListener接收到消息：消息a3
BListener接收到消息：消息ab
上一篇下一篇

猜你喜欢

热点阅读