rabbitmq-交换机

1、交换机是干什么的?
消息(message)由client发送,rabbitmq接收到消息之后通过交换机转发到对应的队列上面。worker会从队列中获取未被读取的数据处理。

1、交换机的种类
rabbitmq包含四种不同的交换机类型:

  • direct exchange:直连交换机,转发消息到routigkey指定的队列
  • fanout exchange:扇形交换机,转发消息到所有绑定队列(速度最快)
  • topic exchange:主题交换机,按规则转发消息(最灵活)
  • headers exchange:首部交换机 (不常用)

1、direct exchange(直连交换机)

直连交换机是一种带路由功能的交换机,根据消息携带的路由键将消息投递给对应队列。
一个队列通过routing_key(路由键)与一个交换机绑定,当消息被发送的时候,需要指定一个routing_key,这个消息被送达交换机的时候,就会被交换机送到指定的队列里面去。
类似一对一的关系!

1.1 springboot的简单实现

创建2个springboot项目,一个 rabbitmq-provider (生产者),一个rabbitmq-consumer(消费者)。

1、首先创建 rabbitmq-provider

pom.xml添加依赖:

	<!--rabbitmq-->
 	 <dependency>
         <groupid>org.springframework.boot</groupid>
         <artifactid>spring-boot-starter-amqp</artifactid>
     </dependency>

配置application.yml:

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");
    }   
}

编写controller层

@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";
    }
}

把rabbitmq-provider项目运行,调用下接口:

因为我们目前还没弄消费者 rabbitmq-consumer,消息没有被消费的,我们去rabbitmq管理页面看看,是否推送成功:

消息已经推送到rabbitmq服务器上面了。

2、创建rabbitmq-consumer项目:
pom.xml和application.yml跟上边一样(把端口号和应用名换了即可)

创建消息接收监听类,directreceiver01.java:

@component
@rabbitlistener(queues = "testdirectqueue")//监听的队列名称 testdirectqueue
public class directreceiver01 {
    @rabbithandler
    public void process(map testmessage) {
        system.out.println("directreceiver01消费者收到的消息: " + testmessage.tostring());
    }
}

启动消费者项目:
运行结果如下

两个消息直接被消费了。然后可以再继续调用rabbitmq-provider项目的推送消息接口,可以看到消费者即时消费消息!

扩展
直连交换机既然是一对一,那如果咱们配置多台监听绑定到同一个直连交互的同一个队列,会怎么样?
直接复制两份directreceiver.java,改下打印内容!

重新启动项目!在生产者多发送几次消息。运行结果如下:

可以看到是实现了轮询的方式对消息进行消费,而且不存在重复消费。

2、topic exchange(主题交换机)

直连交换机的routing_key方案非常简单,但是它是一对一的关系,那么我们需要一对多呢?希望一条消息发送给多个队列。所以rabbitmq提供了一种主题交换机,发送到主题交换机上的消息需要携带指定规则的routing_key,主题交换机会根据这个规则将数据发送到对应的(多个)队列上。

主题交换机的routing_key需要有一定的规则,交换机和队列的binding_key需要采用 *.# 的格式,每个部分用.分开,其中:
*表示一个单词
#表示任意数量(零个或多个)单词。(例如:topic.man)

1、在rabbitmq-provider项目里面创建topicrabbitconfig.java:

@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.#");
      }
}

编写controller层

@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";
    }
}

不要急着去运行,先去编写消费者,创建两个接口

topicmanreceiver :

@component
@rabbitlistener(queues = "topic.man")
public class topicmanreceiver {
    @rabbithandler
    public void process(map testmessage) {
        system.out.println("topicmanreceiver消费者收到消息  : " + testmessage.tostring());
    }
}

topicwomanreceiver :

@component
@rabbitlistener(queues = "topic.woman")
public class topicwomanreceiver {
    @rabbithandler
    public void process(map testmessage) {
        system.out.println("topicwomanreceiver消费者收到消息  : " + testmessage.tostring());
    }
}

然后把rabbitmq-provider,rabbitmq-consumer两个项目都跑起来,先调用/sendtopicmessage1 接口:

然后看消费者rabbitmq-consumer的控制台输出情况:
topicmanreceiver监听队列1,绑定键为:topic.man
topictotalreceiver监听队列2,绑定键为:topic.#
而当前推送的消息,携带的路由键为:topic.man

所以可以看到两个监听消费者receiver都成功消费到了消息,因为这两个recevier监听的队列的绑定键都能与这条消息携带的路由键匹配上。

用/sendtopicmessage2接口:

然后看消费者rabbitmq-consumer的控制台输出情况:
topicmanreceiver监听队列1,绑定键为:topic.man
topictotalreceiver监听队列2,绑定键为:topic.#
而当前推送的消息,携带的路由键为:topic.woman

所以可以看到两个监听消费者,只有topicwomanreceiver 成功消费到了消息,

3、fanout exchange(扇形交换机)

扇形交换机是最基本的交换机类型,它做的事情很简单–广播信息。fanout交换机会把接收到的消息全部转发到绑定的队列上。因为广播不需要“思考”,所以fanout交换机是四种交换机中速度最快的。

同样地,先在rabbitmq-provider项目上创建fanoutrabbitconfig.java:

@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());
    }
}

编写controller层:

@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";
    }
}

接着在rabbitmq-consumer项目里加上消息消费类,

fanoutreceivera:

@component
@rabbitlistener(queues = "fanout.a")
public class fanoutreceivera {
    @rabbithandler
    public void process(map testmessage) {
        system.out.println("fanoutreceiverc消费者收到消息  : " +testmessage.tostring());
    }
}

fanoutreceiverb

@component
@rabbitlistener(queues = "fanout.b")
public class fanoutreceiverb{
    @rabbithandler
    public void process(map testmessage) {
        system.out.println("fanoutreceiverc消费者收到消息  : " +testmessage.tostring());
    }
}

fanoutreceiverc :

@component
@rabbitlistener(queues = "fanout.c")
public class fanoutreceiverc {
    @rabbithandler
    public void process(map testmessage) {
        system.out.println("fanoutreceiverc消费者收到消息  : " +testmessage.tostring());
    }
}

最后将rabbitmq-provider和rabbitmq-consumer项目都跑起来,调用下接口/sendfanoutmessage,运行结果如下:

可以看到只要发送到 fanoutexchange 这个扇型交换机的消息, 三个队列都绑定这个交换机,所以三个消息接收类都监听到了这条消息。

4、消息确认(发送确认与接收确认) 4.1发送确认

在rabbitmq-provider项目中
编写配置文件:

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

配置相关的消息确认回调函数,rabbitconfig.java:

/**
 * 配置相关的消息确认回调函数
 */
@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;
    }
}

confirmcallback
confirmcallback是一个回调接口,就是只确认是否正确到达 exchange 中。
returncallback
通过实现 returncallback 接口,启动消息失败返回,此接口是在交换器路由不到队列时触发回调,该方法可以不使用,因为交换器和队列是在代码里绑定的,如果存在绑定队列失败,那就是你代码写错了。

从总体的情况分析,推送消息存在四种情况:

1、消息推送到server,但是在server里找不到交换机
2、消息推送到server,找到交换机了,但是没找到队列
3、消息推送到sever,交换机和队列啥都没找到
4、消息推送成功

1、消息推送到server,但是在server里找不到交换机
写个测试接口,把消息推送到名为‘non-existent-exchange’的交换机上(这个交换机是没有创建没有配置的):

 @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";
    }

调用接口,查看rabbitmq-provuder项目的控制台输出情况

因为这个交换机没有创建,消息不能到达到达 exchange 中。所以会触发confirmcallback这个回调函数

结论: 只会触发confirmcallback。

2、消息推送到server,找到交换机了,但是没找到队列
新增一个交换机,但是不给这个交换机绑定队列,在directrabitconfig里面新增一个直连交换机,不做任何绑定配置操作:

 @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";
    }

调用方法,查看运行结果:

可以看到两个回调函数都被调用了;
这种情况下,消息是推送成功到服务器了的,所以confirmcallback对消息确认情况是true;
而在retruncallback回调函数的打印参数里面可以看到,消息是推送到了交换机成功了,但是在路由分发给队列的时候,找不到队列,所以报了错误 no_route。
结论: 两个回调函数都会触发!

3、消息推送到sever,交换机和队列啥都没找到
结论: 触发的是 confirmcallback 回调函数。

4、消息推送成功
结论: 触发的是 confirmcallback 回调函数。

总结: confirmcallback回调函数总是会被触发,确认情况true和false是有没有找到交换机,false的话打印原因。returncallback是交换机路由不到队列是回调,也就是你交换机没有绑定队列,这个回调函数可以不写,绑定失败的话就好好看看代码哪里写错了吧。

以上是生产者推送消息的消息确认 回调函数的使用介绍!

4.2 接受确认

消费者确认发生在监听队列的消费者处理业务失败,如,发生了异常,不符合要求的数据……,这些场景我们就需要手动处理,比如重新发送或者丢弃。

消息确认模式有:

  • acknowledgemode.none:自动确认
  • acknowledgemode.auto:根据情况确认
  • acknowledgemode.manual:手动确认

参考:

4.1.1 自动确认

修改消费者配置文件

rabbitmq:
  host: 192.168.152.173
  port: 5672
  username: guest
  password: guest
  listener:
    direct:
      acknowledge-mode: auto   # manual:手动

在消费者项目里,
新建messagelistenerconfig.java上添加代码相关的配置代码:

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;
    }
}

对应的手动确认消息监听类,myackreceiver.java(手动确认模式需要实现 channelawaremessagelistener):

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;
    }
}

需要注意的 basicack 方法需要传递两个参数

  • deliverytag(唯一标识 id):当一个消费者向 rabbitmq 注册后,会建立起一个 channel ,rabbitmq 会用 basic.deliver 方法向消费者推送消息,这个方法携带了一个 delivery tag, 它代表了 rabbitmq 向该 channel 投递的这条消息的唯一标识 id,是一个单调递增的正整数,delivery tag 的范围仅限于 channel
  • multiple:为了减少网络流量,手动确认可以被批处理,当该参数为 true 时,则可以一次性确认 delivery_tag 小于等于传入值的所有消息

basicnack方法需要传递三个参数

  • deliverytag(唯一标识 id):上面已经解释了。
  • multiple:上面已经解释了。
  • requeue: true :重回队列,false :丢弃,我们在nack方法中必须设置 false,否则重发没有意义。

basicreject方法需要传递两个参数

  • deliverytag(唯一标识 id):上面已经解释了。
  • requeue:上面已经解释了,在reject方法里必须设置true。

总结

  1. basic.ack用于肯定确认
  2. basic.nack用于否定确认(注意:这是amqp 0-9-1的rabbitmq扩展)
  3. basic.reject用于否定确认,但与basic.nack相比有一个限制:一次只能拒绝单条消息

消费者端以上的3个方法都表示消息已经被正确投递,但是basic.ack表示消息已经被正确处理。
而basic.nack,basic.reject表示没有被正确处理:

这时,先调用接口,给直连交换机testdirectexchange 的队列testdirectqueue 推送一条消息,可以看到监听器正常消费了下来:

扩展1
除了直连交换机的队列testdirectqueue需要变成手动确认以外,我们还需要将一个其他的队列或者多个队列也变成手动确认,而且不同队列实现不同的业务处理。
那么我们需要做的第一步,往simplemessagelistenercontainer里添加多个队列

myackreceiver.java 就可以同时将上面设置到的队列的消息都消费下来。

@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;
    }
}

分别调用这两个队列相关接口,运行结果如下:

扩展2: 手动确认的另一种写法:
我们在实际工作中可能不会这么写,一般会写在server层,自己定义一个接口:
rabbitconsumerservice

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;
    }
}

@component这个理论上要换成@service,但是由于我没有添加日志,所以换成@service会报错,这里就是提供一个思路,运行结果如下:

附:
rabbitmq在application.yml文件中的其他配置

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

解释:mandatory 设置为 true 则会监听器会接受到路由不可达的消息,然后处理。如果设置为 false,broker 将会自动删除该消息。

参考https://blog.csdn.net/qq_35387940/article/details/100514134

到此这篇关于rabbitmq交换机与springboot整合的简单实现的文章就介绍到这了,更多相关rabbitmq springboot整合内容请搜索www.887551.com以前的文章或继续浏览下面的相关文章希望大家以后多多支持www.887551.com!