RabbitMQ(4) TopicExchange

2022-03-16 00:58:22

topic 是RabbitMQ中最灵活的一种方式，可以根据routing_key*的绑定不同的队列

生产者工程

package com.example.demo.rabbitMq.exchange.topic;

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.Queue;

import org.springframework.amqp.core.TopicExchange;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class TopicRabbitConfig {

    public static final String TOPIC_MESSAGE = "topic.message";

    public static final String TOPIC_MESSAGE_S = "topic.messages";

    public static final String USER_MESSAGE = "user.message";

    /**

     * 武器库

     */

    public static final String ARM_QUEUE = "arm.queue";

    @Bean

    public Queue queueTopicMessage() {

        return new Queue(TopicRabbitConfig.TOPIC_MESSAGE);

    }

    @Bean

    public Queue queueTopicMessages() {

        return new Queue(TopicRabbitConfig.TOPIC_MESSAGE_S);

    }

    @Bean

    public Queue queueUserMessage() {

        return new Queue(TopicRabbitConfig.USER_MESSAGE);

    }

    @Bean

    public Queue queueArm() {

        return new Queue(TopicRabbitConfig.ARM_QUEUE);

    }

    @Bean

    TopicExchange exchange() {

        return new TopicExchange("topicExchange");

    }

    @Bean

    Binding bindingExchangeMessage(Queue queueTopicMessage, TopicExchange exchange) {

        //所有匹配routingKey=topic.message的消息，将放入Queue[name="topic.message"]

        return BindingBuilder.bind(queueTopicMessage).to(exchange).with("topic.message");

    }

    @Bean

    Binding bindingExchangeMessages(Queue queueTopicMessages, TopicExchange exchange) {

        //所有匹配routingKey=topic.# 的消息，将放入Queue[name="topic.messages"]

        return BindingBuilder.bind(queueTopicMessages).to(exchange).with("topic.#");

    }

    @Bean

    Binding bindingExchangeUserMessage(Queue queueUserMessage, TopicExchange exchange) {

        ///所有匹配routingKey=user.# 的消息，将放入Queue[name="user.messages"]

        return BindingBuilder.bind(queueUserMessage).to(exchange).with("user.#");

    }

    @Bean

    Binding bindingExchangeArm(Queue queueArm, TopicExchange exchange) {

        return BindingBuilder.bind(queueArm).to(exchange).with("arm.#");

    }

}

发送消息

package com.example.demo.rabbitMq.exchange.topic;

import com.example.demo.dto.User;

import org.springframework.amqp.core.AmqpTemplate;

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

import org.springframework.stereotype.Component;

@Component

public class TopicSender {

    @Autowired

    private AmqpTemplate rabbitTemplate;

    public void send1() {

        User user = new User();

        user.setUserName("Sender1.....");

        user.setMobile("1111111111");

        rabbitTemplate.convertAndSend("topicExchange","topic.message",user);

    }

    public void send2() {

        User user = new User();

        user.setUserName("Sender2.....");

        user.setMobile("2222222");

        rabbitTemplate.convertAndSend("topicExchange","topic.messages",user);

    }

    public void send3() {

        User user = new User();

        user.setUserName("Sender3.....");

        user.setMobile("33333");

        rabbitTemplate.convertAndSend("topicExchange","user.message",user);

    }

}

消费者工程

package com.example.demo.rabbitMq.exchange.topic;

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.Queue;

import org.springframework.amqp.core.TopicExchange;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class TopicRabbitConstant {

    public static final String TOPIC_MESSAGE = "topic.message";

    public static final String TOPIC_MESSAGE_S = "topic.messages";

    public static final String USER_MESSAGE = "user.message";

}

package com.example.demo.rabbitMq.exchange.topic;

import com.example.demo.dto.User;

import com.example.demo.utils.Base64Utils;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

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

import org.springframework.stereotype.Component;

import java.io.IOException;

@Component

@RabbitListener(queues = TopicRabbitConstant.TOPIC_MESSAGE)

public class TopicReceiver1 {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired

    private AmqpTemplate rabbitTemplate;

    @RabbitHandler

    public void process(User user) {

        System.out.println("Receiver1  : " + user);

    }

    public void rev1(){

        //手动去获取消息

        logger.info("获取Queue[topic.message]消息>>>");

        Message mesg = rabbitTemplate.receive("topic.message");

        System.out.println(mesg);

        if(null != mesg){

            byte[] body = mesg.getBody();

            try {

                User u = (User) Base64Utils.byteToObj(body);

                //获取字符串数据

                System.out.println(u);

            } catch (IOException e) {

                e.printStackTrace();

            } catch (ClassNotFoundException e) {

                e.printStackTrace();

            }

        }

    }

}

测试：

启动消费者工程，生产者，执行如下方法

    @Test

    public void send1() throws Exception {

        //会匹配到topic.#和topic.message 两个Receiver都可以收到消息

        for (int i = 0, size = 10; i < size; i++) {

            topicSender.send1();

        }

    }

也可以不用监听的方式，手动自主获取队列消息，如消费工程：

例如生产者工程TopicRabbitConfig.java添加武器队列：

    /**

     * 武器库

     */

    public static final String ARM_QUEUE = "arm.queue";

    @Bean

    public Queue queueArm() {

        return new Queue(TopicRabbitConfig.ARM_QUEUE);

    }

     @Bean

    Binding bindingExchangeArm(Queue queueArm, TopicExchange exchange) {

        return BindingBuilder.bind(queueArm).to(exchange).with("arm.#");

    }

生产武器：

 public void send4() {

        //生产一批武器

        List<String> list = new ArrayList<String>();

        list.add("手枪");

        list.add("步枪");

        list.add("机枪");

        rabbitTemplate.convertAndSend("topicExchange","arm.gun",list);

    }

    @Test

    public void send4() throws Exception {

        topicSender.send4();

    }

消费者：

package com.example.demo.rabbitMq;

import com.example.demo.dto.User;

import com.example.demo.rabbitMq.exchange.topic.TopicReceiver1;

import com.example.demo.utils.Base64Utils;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.amqp.core.Message;

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

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;

import java.io.IOException;

import java.util.List;

@SpringBootTest

@RunWith(SpringRunner.class)

public class RabbitMqRevTest {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired

    private AmqpTemplate rabbitTemplate;

    @Test

    public void topicRev1(){

        rev1();

    }

    public void rev1(){

        //手动去获取消息

        logger.info("获取Queue[arm.gun]消息>>>");

        Message mesg = rabbitTemplate.receive("arm.queue");

        System.out.println(mesg);

        if(null != mesg){

            byte[] body = mesg.getBody();

            try {

                List u = (List) Base64Utils.byteToObj(body);

                //获取字符串数据

                System.out.println(u);

            } catch (IOException e) {

                e.printStackTrace();

            } catch (ClassNotFoundException e) {

                e.printStackTrace();

            }

        }

    }

}

测试：

样例代码：

https://github.com/xiaozhuanfeng?tab=repositories

码农公寓

相关文章