Direct交换器(发布与订阅)代码实现

桃扇骨 2023-07-22 12:30 12阅读 0赞

1.Direct交换器(发布与订阅)

一.需求图：

二.代码实现：  
步骤1：创建maven项目(springboot)mq-direct-provider、mq-direct-consumer

步骤2：修改全局配置文件  
mq-direct-provider：  
\#设置交换器的名称  
mq.config.exchange=log.direct  
\#info 路由键  
mq.config.queue.info.routing.key=log.info.routing.key  
\#error 路由键  
mq.config.queue.error.routing.key=log.error.routing.key  
mq-direct-consumer：  
\#设置交换器的名称  
mq.config.exchange=log.direct  
\#info 队列名称  
mq.config.queue.info=log.info  
\#info 路由键  
mq.config.queue.info.routing.key=log.info.routing.key  
\#error 队列名称  
mq.config.queue.error=log.error  
\#error 路由键  
mq.config.queue.error.routing.key=log.error.routing.key

步骤2：代码编写  
mq-direct-provider：  
/\*\*

*  消息发送者  
    \*\*/  
    @Component  
    public class Sender \{  
    @Autowired  
    private AmqpTemplate rabbitAmqpTemplate;  
    // exchange 交换器名称  
    @Value(" m q . c o n f i g . e x c h a n g e " ) p r i v a t e S t r i n g e x c h a n g e ; / / r o u t i n g k e y 路 由 键 @ V a l u e ( " \{mq.config.exchange\}") private String exchange; // routingkey 路由键 @Value(" mq.config.exchange")privateStringexchange;//routingkey路由键@Value("\{mq.config.queue.error.routing.key\}")  
    private String routingkey1;  
    @Value("$\{mq.config.queue.info.routing.key\}")  
    private String routingkey2;

/*发送消息的方法 */
    public void send(String msg) {
     // 向消息队列发送消息
     // 参数一：交换器名称 ， 参数二：路由键，   参数：消息
     this.rabbitAmqpTemplate.convertAndSend(this.exchange, this.routingkey1, msg);
     this.rabbitAmqpTemplate.convertAndSend(this.exchange, this.routingkey2, msg);
    }

\}  
/\*\*

*  测试代码  
    **/  
    @RunWith(SpringRunner.class)  
    @SpringBootTest(classes=AppStart.class)  
    public class RabbitmqTest \{  
    @Autowired  
    private Sender sender;  
    /*测试消息队列*/  
    @Test  
    public void test1() throws Exception\{  
    while(true) \{  
    Thread.sleep(2000);  
    this.sender.send(“Hello RabbitMQ”);  
    \}  
    \}  
    \}  
    mq-direct-consumer：  
    /**
 *  消息接收者
 *  @RabbitListener bindings:绑定队列
 *  @QueueBinding value:绑定队列的名称 exchange:配置交换器
 *  @Queue value:配置队列名称 autoDelete:是否是一个可删除的临时队列
 *  @Exchange value:为交换器起个名称 type:指定具体的交换器类型  
    */  
    @Component  
    @RabbitListener(bindings = @QueueBinding(  
    value = @Queue(value = “ m q . c o n f i g . q u e u e . e r r o r " , a u t o D e l e t e = " t r u e " ) , / / 消 息 队 列 名 称 绑 定 e x c h a n g e = @ E x c h a n g e ( v a l u e = " \{mq.config.queue.error\}", autoDelete = "true"), //消息队列名称绑定 exchange = @Exchange(value = " mq.config.queue.error",autoDelete="true"),//消息队列名称绑定exchange=@Exchange(value="\{mq.config.exchange\}”, type = ExchangeTypes.DIRECT), //交换器名称绑定  
    key = “$\{mq.config.queue.error.routing.key\}” //路由键，把消息队列和交换器建立连接  
    ))  
    public class ErrorReceiver \{  
    /*\*
    
     *  接收消息的方法,采用消息队列监听机制@RabbitListener  
        \*/  
        @RabbitHandler  
        public void process(String msg) \{System.out.println("Error…receiver: " + msg);\}  
        \}

/\*\*

*  消息接收者
 *  @RabbitListener bindings:绑定队列
 *  @QueueBinding value:绑定队列的名称 exchange:配置交换器
 *  @Queue value:配置队列名称 autoDelete:是否是一个可删除的临时队列
 *  @Exchange value:为交换器起个名称 type:指定具体的交换器类型  
    */  
    @Component  
    @RabbitListener(bindings = @QueueBinding(  
    value = @Queue(value = “ m q . c o n f i g . q u e u e . i n f o " , a u t o D e l e t e = " t r u e " ) , / / 消 息 队 列 名 称 绑 定 e x c h a n g e = @ E x c h a n g e ( v a l u e = " \{mq.config.queue.info\}", autoDelete = "true"), //消息队列名称绑定 exchange = @Exchange(value = " mq.config.queue.info",autoDelete="true"),//消息队列名称绑定exchange=@Exchange(value="\{mq.config.exchange\}”, type = ExchangeTypes.DIRECT), //交换器名称绑定  
    key = “$\{mq.config.queue.info.routing.key\}” //路由键，把消息队列和交换器建立连接  
    ))  
    public class InfoReceiver \{  
    /*\*
    
     *  接收消息的方法,采用消息队列监听机制@RabbitListener  
        \*/  
        @RabbitHandler  
        public void process(String msg) \{System.out.println("Info…receiver: " + msg);\}  
        \}