RabbitMQ交换机与Springboot整合的简单实现

<!--rabbitmq-->

  <dependency>

      <groupid>org.springframework.boot</groupid>

      <artifactid>spring-boot-starter-amqp</artifactid>

  </dependency>

server:

   port: 8090

spring:

   application:

     name: rabbitmq-provider

   rabbitmq:

     host: 192.168 . 152.173

     port: 5672      #注意 15672 是web页面端口，消息发送端口为 5672

     username: guest

     password: guest

@configuration

public class directrabbitconfig {

     /**

      * 创建消息队列 起名：testdirectqueue

      * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效

      * exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable

      * autodelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。

      * 一般设置一下队列的持久化就好,其余两个就是默认false

      * @return new queue(name,durable,exclusive,autodelete)

      */

     @bean

     public queue testdirectqueue() {

         return new queue( "testdirectqueue" , true );

     }

     /**

      * direct交换机 起名：testdirectexchange

      */

     @bean

     directexchange testdirectexchange() {

         return new directexchange( "testdirectexchange" , true , false );

     }

     /**

      * 绑定

      * 将队列和交换机绑定, 并设置用于匹配键（路由键）：testdirectrouting

      */

     @bean

     binding bindingdirect() {

         return bindingbuilder.bind(testdirectqueue()).to(testdirectexchange()).with( "testdirectrouting" );

     }  

}

@restcontroller

@requestmapping ( "/rabbit" )

public class sendmessagecontroller {

     @autowired

     private rabbittemplate rabbittemplate;  //使用rabbittemplate,这提供了接收/发送等等方法

     @requestmapping ( "/send" )

     public string senddirectmessage() {

         string messageid = string.valueof(uuid.randomuuid());

         string messagedata = "league of legends never dies！" ;

         string createtime = localdatetime.now().format(datetimeformatter.ofpattern( "yyyy-mm-dd hh:mm:ss" ));

         map<string,object> map= new hashmap<>();

         map.put( "messageid" ,messageid);

         map.put( "messagedata" ,messagedata);

         map.put( "createtime" ,createtime);

         //将消息携带绑定键值：testdirectrouting 发送到交换机testdirectexchange

         rabbittemplate.convertandsend( "testdirectexchange" , "testdirectrouting" , map);

         return "ok" ;

     }

}

@component

@rabbitlistener (queues = "testdirectqueue" ) //监听的队列名称 testdirectqueue

public class directreceiver01 {

     @rabbithandler

     public void process(map testmessage) {

         system.out.println( "directreceiver01消费者收到的消息: " + testmessage.tostring());

     }

}

@configuration

public class topicrabbitconfig {

     //绑定键

       public final static string man = "topic.man" ;

       public final static string woman = "topic.woman" ;

       @bean

       public queue firstqueue() {

           return new queue(topicrabbitconfig.man);

       }

       @bean

       public queue secondqueue() {

           return new queue(topicrabbitconfig.woman);

       }

       @bean

       topicexchange exchange() {

           return new topicexchange( "topicexchange" );

       }

       //将firstqueue和topicexchange绑定,而且绑定的键值为topic.man

       //这样只要是消息携带的路由键是topic.man,才会分发到该队列

       @bean

       binding bindingexchangemessage() {

           return bindingbuilder.bind(firstqueue()).to(exchange()).with(man);

       }

       //将secondqueue和topicexchange绑定,而且绑定的键值为用上通配路由键规则topic.#

       // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列

       @bean

       binding bindingexchangemessage2() {

           return bindingbuilder.bind(secondqueue()).to(exchange()).with( "topic.#" );

       }

}

@requestmapping ( "rabbit" )

@restcontroller

public class topicsendcontroller {

     @autowired

     private rabbittemplate template;

     @getmapping ( "/sendtopicmessage1" )

     public string sendtopicmessage1() {

         string messageid = string.valueof(uuid.randomuuid());

         string messagedata = "message: m a n " ;

         string createtime = localdatetime.now().format(datetimeformatter.ofpattern( "yyyy-mm-dd hh:mm:ss" ));

         map<string, object> manmap = new hashmap<>();

         manmap.put( "messageid" , messageid);

         manmap.put( "messagedata" , messagedata);

         manmap.put( "createtime" , createtime);

         template.convertandsend( "topicexchange" , "topic.man" , manmap);

         return "ok" ;

     }

     @getmapping ( "/sendtopicmessage2" )

     public string sendtopicmessage2() {

         string messageid = string.valueof(uuid.randomuuid());

         string messagedata = "message: woman is all " ;

         string createtime = localdatetime.now().format(datetimeformatter.ofpattern( "yyyy-mm-dd hh:mm:ss" ));

         map<string, object> womanmap = new hashmap<>();

         womanmap.put( "messageid" , messageid);

         womanmap.put( "messagedata" , messagedata);

         womanmap.put( "createtime" , createtime);

         template.convertandsend( "topicexchange" , "topic.woman" , womanmap);

         return "ok" ;

     }

}

@component

@rabbitlistener (queues = "topic.man" )

public class topicmanreceiver {

     @rabbithandler

     public void process(map testmessage) {

         system.out.println( "topicmanreceiver消费者收到消息  : " + testmessage.tostring());

     }

}

@component

@rabbitlistener (queues = "topic.woman" )

public class topicwomanreceiver {

     @rabbithandler

     public void process(map testmessage) {

         system.out.println( "topicwomanreceiver消费者收到消息  : " + testmessage.tostring());

     }

}

@configuration

public class fanoutrabbitconfig {

     /**

      * 创建三个队列 ：fanout.a   fanout.b  fanout.c

      * 将三个队列都绑定在交换机 fanoutexchange 上

      * 因为是扇型交换机, 路由键无需配置,配置也不起作用

      */

     @bean

     public queue queuea() {

         return new queue( "fanout.a" );

     }

     @bean

     public queue queueb() {

         return new queue( "fanout.b" );

     }

     @bean

     public queue queuec() {

         return new queue( "fanout.c" );

     }

     @bean

     fanoutexchange fanoutexchange() {

         return new fanoutexchange( "fanoutexchange" );

     }

     @bean

     binding bindingexchangea() {

         return bindingbuilder.bind(queuea()).to(fanoutexchange());

     }

     @bean

     binding bindingexchangeb() {

         return bindingbuilder.bind(queueb()).to(fanoutexchange());

     }

     @bean

     binding bindingexchangec() {

         return bindingbuilder.bind(queuec()).to(fanoutexchange());

     }

}

@restcontroller

@requestmapping ( "rabbit" )

public class fanoutcontroller {

     @autowired

     private rabbittemplate template;

     @getmapping ( "/sendfanoutmessage" )

     public string sendfanoutmessage() {

         string messageid = string.valueof(uuid.randomuuid());

         string messagedata = "message: testfanoutmessage " ;

         string createtime = localdatetime.now().format(datetimeformatter.ofpattern( "yyyy-mm-dd hh:mm:ss" ));

         map<string, object> map = new hashmap<>();

         map.put( "messageid" , messageid);

         map.put( "messagedata" , messagedata);

         map.put( "createtime" , createtime);

         template.convertandsend( "fanoutexchange" , null , map);

         return "ok" ;

     }

}

@component

@rabbitlistener (queues = "fanout.a" )

public class fanoutreceivera {

     @rabbithandler

     public void process(map testmessage) {

         system.out.println( "fanoutreceiverc消费者收到消息  : " +testmessage.tostring());

     }

}

@component

@rabbitlistener (queues = "fanout.b" )

public class fanoutreceiverb{

     @rabbithandler

     public void process(map testmessage) {

         system.out.println( "fanoutreceiverc消费者收到消息  : " +testmessage.tostring());

     }

}

@component

@rabbitlistener (queues = "fanout.c" )

public class fanoutreceiverc {

     @rabbithandler

     public void process(map testmessage) {

         system.out.println( "fanoutreceiverc消费者收到消息  : " +testmessage.tostring());

     }

}

server:

   port: 8090

spring:

   application:

     name: rabbitmq-consumer

   rabbitmq:

     host: 192.168 . 152.173

     port: 5672

     username: guest

     password: guest

     #确认消息已发送到交换机(exchange)

     publisher-confirm-type: correlated

     #确认消息已发送到队列(queue)

     publisher-returns: true

/**

  * 配置相关的消息确认回调函数

  */

@configuration

public class rabbitconfig {

     @bean

     public rabbittemplate createrabbittemplate(connectionfactory connectionfactory) {

         rabbittemplate rabbittemplate = new rabbittemplate();

         rabbittemplate.setconnectionfactory(connectionfactory);

         //设置开启mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数

         rabbittemplate.setmandatory( true );

         //确认回调

         rabbittemplate.setconfirmcallback( new rabbittemplate.confirmcallback() {

             @override

             public void confirm(correlationdata correlationdata, boolean condition, string cause) {

                 system.out.println( "confirmcallback:     " + "相关数据：" + correlationdata);

                 system.out.println( "confirmcallback:     " + "确认情况：" + condition);

                 system.out.println( "confirmcallback:     " + "原因：" + cause);

             }

         });

         //设置返回消息的回调

         rabbittemplate.setreturncallback( new rabbittemplate.returncallback() {

             @override

             public void returnedmessage(message message, int replycode, string replytext, string exchange, string routingkey) {

                 system.out.println( "returncallback:     " + "消息：" + message);

                 system.out.println( "returncallback:     " + "回应码：" + replycode);

                 system.out.println( "returncallback:     " + "回应信息：" + replytext);

                 system.out.println( "returncallback:     " + "交换机：" + exchange);

                 system.out.println( "returncallback:     " + "路由键：" + routingkey);

             }

         });

         return rabbittemplate;

     }

}

@getmapping ( "/testmessageack" )

    public string testmessageack() {

        string messageid = string.valueof(uuid.randomuuid());

        string messagedata = "message: non-existent-exchange test message " ;

        string createtime = localdatetime.now().format(datetimeformatter.ofpattern( "yyyy-mm-dd hh:mm:ss" ));

        map<string, object> map = new hashmap<>();

        map.put( "messageid" , messageid);

        map.put( "messagedata" , messagedata);

        map.put( "createtime" , createtime);

        //non-existent-exchange这个交换机没有创建和配置

        template.convertandsend( "non-existent-exchange" , "testdirectrouting" , map);

        return "ok" ;

    }

@bean

    directexchange lonelydirectexchange() {

        return new directexchange( "lonelydirectexchange" );

    }

@getmapping ( "/testmessageack2" )

    public string testmessageack2() {

        string messageid = string.valueof(uuid.randomuuid());

        string messagedata = "message: lonelydirectexchange test message " ;

        string createtime = localdatetime.now().format(datetimeformatter.ofpattern( "yyyy-mm-dd hh:mm:ss" ));

        map<string, object> map = new hashmap<>();

        map.put( "messageid" , messageid);

        map.put( "messagedata" , messagedata);

        map.put( "createtime" , createtime);

        rabbittemplate.convertandsend( "lonelydirectexchange" , "testdirectrouting" , map);

        return "ok" ;

    }

rabbitmq:

   host: 192.168 . 152.173

   port: 5672

   username: guest

   password: guest

   listener:

     direct:

       acknowledge-mode: auto   # manual：手动

package com.zsn.demo.config;

import com.zsn.demo.resceiver.myackreceiver;

import org.springframework.amqp.core.acknowledgemode;

import org.springframework.amqp.rabbit.connection.cachingconnectionfactory;

import org.springframework.amqp.rabbit.listener.simplemessagelistenercontainer;

import org.springframework.beans.factory.annotation.autowired;

import org.springframework.context.annotation.bean;

import org.springframework.context.annotation.configuration;

/**

  * @author： zhouzhou

  * @date：2021/7/22 9:49

  */

@configuration

public class messagelistenerconfig {

     @autowired

     private cachingconnectionfactory connectionfactory;

     @autowired

     private myackreceiver myackreceiver; //消息接收处理类

     @bean

     public simplemessagelistenercontainer simplemessagelistenercontainer() {

         simplemessagelistenercontainer container = new simplemessagelistenercontainer(connectionfactory);

         container.setconcurrentconsumers( 1 );

         container.setmaxconcurrentconsumers( 1 );

         container.setacknowledgemode(acknowledgemode.manual); // rabbitmq默认是自动确认，这里改为手动确认消息

         //设置一个队列

         container.setqueuenames( "testdirectqueue" );

         //如果同时设置多个如下： 前提是队列都是必须已经创建存在的

         //  container.setqueuenames("testdirectqueue","testdirectqueue2","testdirectqueue3");

         //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addqueues

         //container.setqueues(new queue("testdirectqueue",true));

         //container.addqueues(new queue("testdirectqueue2",true));

         //container.addqueues(new queue("testdirectqueue3",true));

         container.setmessagelistener(myackreceiver);

         return container;

     }

}

package com.zsn.demo.resceiver;

import com.rabbitmq.client.channel;

import org.springframework.amqp.core.message;

import org.springframework.amqp.rabbit.listener.api.channelawaremessagelistener;

import org.springframework.stereotype.component;

import java.util.hashmap;

import java.util.map;

/**

  * @author： zhouzhou

  * @date：2021/7/22 9:50

  */

@component

public class myackreceiver implements channelawaremessagelistener {

     /**

      * 用于处理接收到的 rabbit 消息的回调。

      * 实现者应该处理指定的信息

      * 通常通过给定的会话发送回复消息

      * @param message the received amqp message (never <code>null</code>)

      * @param channel the underlying rabbit channel (never <code>null</code>)

      * @throws exception any.

      */

     @override

     public void onmessage(message message, channel channel) throws exception {

         long deliverytag = message.getmessageproperties().getdeliverytag();

         try {

             //因为传递消息的时候用的map传递,所以将map从message内取出需要做些处理

             string msg = message.tostring();

             string[] msgarray = msg.split( "'" ); //可以点进message里面看源码,单引号直接的数据就是我们的map消息数据

             map<string, string> msgmap = mapstringtomap(msgarray[ 1 ].trim(), 3 );

             string messageid=msgmap.get( "messageid" );

             string messagedata=msgmap.get( "messagedata" );

             string createtime=msgmap.get( "createtime" );

             system.out.println( "  myackreceiver  messageid:" +messageid+ "  messagedata:" +messagedata+ "  createtime:" +createtime);

             system.out.println( "消费的主题消息来自：" +message.getmessageproperties().getconsumerqueue());

             channel.basicack(deliverytag, true ); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息

//          channel.basicreject(deliverytag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝

         } catch (exception e) {

             channel.basicreject(deliverytag, false );

             e.printstacktrace();

         }

     }

     //{key=value,key=value,key=value} 格式转换成map

     private map<string, string> mapstringtomap(string str, int entrynum ) {

         str = str.substring( 1 , str.length() - 1 );

         string[] strs = str.split( "," ,entrynum);

         map<string, string> map = new hashmap<string, string>();

         for (string string : strs) {

             string key = string.split( "=" )[ 0 ].trim();

             string value = string.split( "=" )[ 1 ];

             map.put(key, value);

         }

         return map;

     }

}

@component

public class myackreceiver implements channelawaremessagelistener {

    /**

      * 用于处理接收到的 rabbit 消息的回调。

      * 实现者应该处理指定的信息

      * 通常通过给定的会话发送回复消息

      * @param message the received amqp message (never <code>null</code>)

      * @param channel the underlying rabbit channel (never <code>null</code>)

      * @throws exception any.

      */

     @override

     public void onmessage(message message, channel channel) throws exception {

         long deliverytag = message.getmessageproperties().getdeliverytag();

         try {

             //因为传递消息的时候用的map传递,所以将map从message内取出需要做些处理

             string msg = message.tostring();

             string[] msgarray = msg.split( "'" ); //可以点进message里面看源码,单引号直接的数据就是我们的map消息数据

             map<string, string> msgmap = mapstringtomap(msgarray[ 1 ].trim(), 3 );

             string messageid=msgmap.get( "messageid" );

             string messagedata=msgmap.get( "messagedata" );

             string createtime=msgmap.get( "createtime" );

             if ( "testdirectqueue" .equals(message.getmessageproperties().getconsumerqueue())){

                 system.out.println( "消费的消息来自的队列名为：" +message.getmessageproperties().getconsumerqueue());

                 system.out.println( "消息成功消费到  messageid:" +messageid+ "  messagedata:" +messagedata+ "  createtime:" +createtime);

                 system.out.println( "执行testdirectqueue中的消息的业务处理流程......" );

             }

             if ( "fanout.a" .equals(message.getmessageproperties().getconsumerqueue())){

                 system.out.println( "消费的消息来自的队列名为：" +message.getmessageproperties().getconsumerqueue());

                 system.out.println( "消息成功消费到  messageid:" +messageid+ "  messagedata:" +messagedata+ "  createtime:" +createtime);

                 system.out.println( "执行fanout.a中的消息的业务处理流程......" );

             }

             channel.basicack(deliverytag, true );

//          channel.basicreject(deliverytag, true);//为true会重新放回队列

         } catch (exception e) {

             channel.basicreject(deliverytag, false );

             e.printstacktrace();

         }

     }

     //{key=value,key=value,key=value} 格式转换成map

     private map<string, string> mapstringtomap(string str, int ennum) {

         str = str.substring( 1 , str.length() - 1 );

         string[] strs = str.split( "," ,ennum);

         map<string, string> map = new hashmap<string, string>();

         for (string string : strs) {

             string key = string.split( "=" )[ 0 ].trim();

             string value = string.split( "=" )[ 1 ];

             map.put(key, value);

         }

         return map;

     }

}

public interface rabbitconsumerservice {

     void process(channel channel, message message) throws ioexception;

}

@component

public class rabbitconsumerserviceimpl implements rabbitconsumerservice{

     @rabbithandler

     @rabbitlistener (queues = "testdirectqueue" )

     @override

     public void process(channel channel, message message) throws ioexception {

         long deliverytag = message.getmessageproperties().getdeliverytag();

         try {

             //因为传递消息的时候用的map传递,所以将map从message内取出需要做些处理

             string msg = message.tostring();

             string[] msgarray = msg.split( "'" ); //可以点进message里面看源码,单引号直接的数据就是我们的map消息数据

             map<string, string> msgmap = mapstringtomap(msgarray[ 1 ].trim(), 3 );

             string messageid=msgmap.get( "messageid" );

             string messagedata=msgmap.get( "messagedata" );

             string createtime=msgmap.get( "createtime" );

             system.out.println( "  myackreceiver  messageid:" +messageid+ "  messagedata:" +messagedata+ "  createtime:" +createtime);

             system.out.println( "消费的主题消息来自：" +message.getmessageproperties().getconsumerqueue());

             channel.basicack(deliverytag, true ); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息

//          channel.basicreject(deliverytag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝

         } catch (exception e) {

             channel.basicreject(deliverytag, false );

             e.printstacktrace();

         }

     }

     //{key=value,key=value,key=value} 格式转换成map

     private map<string, string> mapstringtomap(string str, int entrynum ) {

         str = str.substring( 1 , str.length() - 1 );

         string[] strs = str.split( "," ,entrynum);

         map<string, string> map = new hashmap<string, string>();

         for (string string : strs) {

             string key = string.split( "=" )[ 0 ].trim();

             string value = string.split( "=" )[ 1 ];

             map.put(key, value);

         }

         return map;

     }

}

rabbitmq:

     addresses: 192.168 . 152.193 : 5672

     username: guest

     password: guest 

     #虚拟host 可以不设置,使用server默认host

     virtual-host: /

     listener:

       simple:

         prefetch: 1                #设置一次处理一个消息

         acknowledge-mode: manual  #设置消费端手动 ack

         concurrency: 3             #设置同时有 3 个消费者消费

     connection-timeout: 500

     # 确认消息是否正确到达queue,如果没有则触发，如果有则不触发

     publisher-returns: true

     #确认消息已发送到交换机(exchange)

     publisher-confirm-type: correlated

     template:

       #设置为 true 后 消费者在消息没有被路由到合适队列情况下会被 return 监听，而不会自动删除

       mandatory: true