SpringBoot使用消息中间件RabbitMQ

Bertha 。 2022-05-19 01:36 352阅读 0赞

首先在docker中安装rabbitmq,pull 带有web界面的

docker pull rabbitmq:3-management

5672为客户端,15672为web界面端口

docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq01 镜像id

简要介绍一下rabbitmq的三个Exchange

Fanout（ 广播）  
绑定queue时，不用设置routkey  
发布消息时，不用设置routkey

Direct （点对点，单播）  
绑定queue时，设置routkey  
发布消息时，设置routkey，单播

Topic  
绑定queue时，使用包含\* 和\#的表达式  
\#代表一个或多个单词  
\*代表一个单词  
发布消息时，绑定queue的表达式中，匹配到的queue 则收到消息

接下来在springboot项目中使用rabbitmq  
在pom.xml中引入依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>

配置

spring.rabbitmq.host=120.87.111.104
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

开启注解

@EnableRabbit
    @SpringBootApplication
    public class RabbitmqApplication { 
    
        public static void main(String[] args) {
            SpringApplication.run(RabbitmqApplication.class, args);
        }
    }

初始化

@Autowired
        AmqpAdmin admin;
    
        //初始化
        @Test
        public void test1(){
            //创建direct
            admin.declareExchange(new DirectExchange("admin.direct"));
            //创建fanout
            admin.declareExchange(new FanoutExchange("admin.fanout"));
            //创建topic
            admin.declareExchange(new TopicExchange("admin.topic"));
            //创建3个消息队列
            admin.declareQueue(new Queue("admin.lhc1"));
            admin.declareQueue(new Queue("admin.lhc2"));
            admin.declareQueue(new Queue("admin.lhc3"));
            //设置绑定关系
            //设置direct与三个消息队列的关系
            admin.declareBinding(new Binding("admin.lhc1", Binding.DestinationType.QUEUE,"admin.direct","admin.lhc1",null));
            admin.declareBinding(new Binding("admin.lhc2", Binding.DestinationType.QUEUE,"admin.direct","admin.lhc2",null));
            admin.declareBinding(new Binding("admin.lhc3", Binding.DestinationType.QUEUE,"admin.direct","admin.lhc3",null));
            //设置fanout与三个消息队列的关系
            admin.declareBinding(new Binding("admin.lhc1",Binding.DestinationType.QUEUE,"admin.fanout","lhc1",null));
            admin.declareBinding(new Binding("admin.lhc2",Binding.DestinationType.QUEUE,"admin.fanout","lhc2",null));
            admin.declareBinding(new Binding("admin.lhc3",Binding.DestinationType.QUEUE,"admin.fanout","lhc3",null));
            //设置topic与三个消息队列的关系
            admin.declareBinding(new Binding("admin.lhc1", Binding.DestinationType.QUEUE,"admin.topic","admin.lhc.com.#",null));
            admin.declareBinding(new Binding("admin.lhc2", Binding.DestinationType.QUEUE,"admin.topic","admin.lhc.#",null));
            admin.declareBinding(new Binding("admin.lhc3", Binding.DestinationType.QUEUE,"admin.topic","admin.lhc.*",null));
        }

设置MessageConverter

@Configuration
    public class RabbitMQConfig { 
    
        //默认使用java的序列化机制，需要修改为Jackson2
        @Bean
        public MessageConverter messageConverter(){
            return new Jackson2JsonMessageConverter();
        }
    }

进行相关测试

//单播
        @Test
        public void test3(){
         User user= new User("1","jackson","123456");
            rabbitTemplate.convertAndSend("admin.direct","admin.lhc1",user);
        }
        //广播
        @Test
        public void test5(){
            User user=  new User("1","jerry","123456");
            //不需要传递routKey，但必需传空的字符串，否则消息队列接收不到
            rabbitTemplate.convertAndSend("admin.fanout","",user);
        }
    
        //topic
        @Test
        public void test6(){
            User user=  new User("1","jack","123456");
            //admin.topic与三个队列的绑定关系如下，故可确定admin.lhc1和admin.lhc2可以收到
            //"admin.lhc.com.#"
            //"admin.lhc.#"
            //"admin.lhc.*"
            rabbitTemplate.convertAndSend("admin.topic","admin.lhc.com.topic",user);
        }
    
        //接收
        @Test
        public void test4(){
            Object o=rabbitTemplate.receiveAndConvert("admin.lhc1");
            System.out.println(o.getClass());
            System.out.println(o);
        }

还可以在Service中进行监听

@Service
    public class UserService { 
    
        @RabbitListener(queues = "admin.lhc2")
        public void receive2(Message message){
            System.out.println(message.getBody());
            System.out.println(message.getMessageProperties());
        }
    
        @RabbitListener(queues = "admin.lhc1")
        //自动类型转化
        public void receive(User user){
            System.out.println(user);
        }
    
    }