集成RabbitMQ+MQ常用操作 - 实践

文章目录

• • 1.环境搭建
  • • • 1.Docker安装RabbitMQ
      • • 1.拉取镜像
        • 2.安装命令
        • 3.开启5672和15672端口
        • 4.登录控制台
      • 2.整合Spring AMQP
      • • 1.sun-common模块下创建新模块
        • 2.引入amqp依赖和fastjson
      • 3.新建一个mq-demo的模块
      • • 1.在sun-frame下创建mq-demo
        • 2.然后在mq-demo下创建生产者和消费者子模块
        • 3.查看是否交给父模块管理了
        • 4.在mq-demo模块引入sun-common-rabbitmq依赖
        • 5.publisher引入sun-common-test依赖
        • 6.将sun-common-rabbitmq clean-install一下
        • 7.给consumer和publisher都创建主类
        • • 1.ConsumerApplication.java
          • 2.PublisherApplication.java
      • 4.测试MQ
      • • 1.application.yml mq的最基本配置
        • 2.consumer
        • • 1.TestConfig.java MQ配置
          • 2.TestConfigListener.java 监听队列
        • 3.publisher
        • • 1.TestConfig.java 测试(注意指定启动类)
          • 2.结果
  • 2.基本交换机
  • • • 1.Fanout
      • • 1.FanoutConfig.java 交换机配置
        • 2.FanoutConfigListener.java 监听者
        • 3.FanoutConfig.java 生产者
      • 2.Direct
      • • 1.DirectConfig.java 交换机配置
        • 2.DirectConfigListener.java 监听者
        • 3.DirectConfig.java 生产者

1.环境搭建

1.Docker安装RabbitMQ

1.拉取镜像

docker pull rabbitmq:3.8-management

2.安装命令

docker run -e RABBITMQ_DEFAULT_USER=sun-e RABBITMQ_DEFAULT_PASS=mq-v mq-plugins:/plugins--name mq--hostname mq-p 15672:15672-p 5672:5672-d 699038cb2b96 # 注意这里是镜像id，需要替换

3.开启5672和15672端口

4.登录控制台

15672端口

2.整合Spring AMQP

1.sun-common模块下创建新模块

2.引入amqp依赖和fastjson

4.0.0com.sunxianshengsun-common1.0-SNAPSHOTsun-common-rabbitmq${children.version}org.springframework.bootspring-boot-starter-amqpcom.alibabafastjson

3.新建一个mq-demo的模块

1.在sun-frame下创建mq-demo

2.然后在mq-demo下创建生产者和消费者子模块

3.查看是否交给父模块管理了

4.在mq-demo模块引入sun-common-rabbitmq依赖

    com.sunxianshengsun-common-rabbitmq1.0-SNAPSHOT

5.publisher引入sun-common-test依赖

    com.sunxianshengsun-common-test1.0-SNAPSHOT

6.将sun-common-rabbitmq clean-install一下

7.给consumer和publisher都创建主类

1.ConsumerApplication.java

package com.sunxiansheng.consumer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
@SpringBootApplication
@ComponentScan("com.sunxiansheng")
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}
}

2.PublisherApplication.java

package com.sunxiansheng.publisher;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class PublisherApplication {public static void main(String[] args) {SpringApplication.run(PublisherApplication.class);}
}

4.测试MQ

1.application.yml mq的最基本配置

spring:# RabbitMQ 配置rabbitmq:# 服务器地址host: ip# 用户名username: sunxiansheng# 密码password: rabbitmq# 虚拟主机virtual-host: /# 端口port: 5672

2.consumer

1.TestConfig.java MQ配置

package com.sunxiansheng.consumer.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/*** Description: 最基本的MQ测试* @Author sun* @Create 2024/8/2 14:34* @Version 1.0*/
@Configuration
public class TestConfig {/*** 创建一个fanout类型的交换机* @return*/@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("fanout.exchange.test");}/*** 创建一个队列* @return*/@Beanpublic Queue fanoutQueueTest() {return new Queue("fanout.queue.test");}/*** 交换机和队列绑定*/@Beanpublic Binding binding() {return BindingBuilder.bind(fanoutQueueTest()).to(fanoutExchange());}
}

2.TestConfigListener.java 监听队列

package com.sunxiansheng.consumer.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/*** Description: 最基本的MQ测试* @Author sun* @Create 2024/8/2 14:34* @Version 1.0*/
@Component
public class TestConfigListener {@RabbitListener(queues = "fanout.queue.test")public void receive(String message) {System.out.println("接收到的消息：" + message);}
}

3.publisher

1.TestConfig.java 测试(注意指定启动类)

package com.sunxiansheng.consumer.config;
import com.sunxiansheng.publisher.PublisherApplication;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
/*** Description: 最基本的MQ测试* @Author sun* @Create 2024/8/2 14:34* @Version 1.0*/
@SpringBootTest(classes = PublisherApplication.class) // 指定启动类
public class TestConfig {@Resourceprivate AmqpTemplate amqpTemplate;@Testpublic void send() {// 交换机String exchange = "fanout.exchange.test";// 路由键String routingKey = "";// 消息String message = "hello fanout";// 发送消息amqpTemplate.convertAndSend(exchange, routingKey, message);}
}

2.结果

2.基本交换机

1.Fanout

1.FanoutConfig.java 交换机配置

package com.sunxiansheng.consumer.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/*** Description: Fanout交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Configuration
public class FanoutConfig {@Beanpublic FanoutExchange fanoutExchange1() {// 创建一个fanout类型的交换机return new FanoutExchange("fanout.exchange");}@Beanpublic Queue fanoutQueue1() {// 创建一个队列return new Queue("fanout.queue1");}@Beanpublic Queue fanoutQueue2() {// 创建一个队列return new Queue("fanout.queue2");}// 两个队列绑定到交换机上@Beanpublic Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange1) {return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange1);}@Beanpublic Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange1) {return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange1);}
}

2.FanoutConfigListener.java 监听者

package com.sunxiansheng.consumer.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/*** Description: Fanout交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Component
public class FanoutConfigListener {@RabbitListener(queues = "fanout.queue1")public void receive1(String message) {System.out.println("fanout.queue1接收到的消息：" + message);}@RabbitListener(queues = "fanout.queue2")public void receive2(String message) {System.out.println("fanout.queue2接收到的消息：" + message);}
}

3.FanoutConfig.java 生产者

package com.sunxiansheng.consumer.config;
import com.sunxiansheng.publisher.PublisherApplication;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
/*** Description: Fanout交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@SpringBootTest(classes = PublisherApplication.class) // 指定启动类
public class FanoutConfig {@Resourceprivate AmqpTemplate amqpTemplate;@Testpublic void send() {// 交换机String exchange = "fanout.exchange";// 路由键String routingKey = "";// 消息String message = "hello fanout";// 发送消息amqpTemplate.convertAndSend(exchange, routingKey, message);}
}

2.Direct

1.DirectConfig.java 交换机配置

package com.sunxiansheng.consumer.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/*** Description: Direct交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Configuration
public class DirectConfig {@Beanpublic DirectExchange directExchange() {// 创建一个direct类型的交换机return new DirectExchange("direct.exchange");}@Beanpublic Queue directQueue1() {// 创建一个队列return new Queue("direct.queue1");}@Beanpublic Queue directQueue2() {// 创建一个队列return new Queue("direct.queue2");}// 两个队列绑定到交换机上，这里需要指定routingKey@Beanpublic Binding bindingDirectQueue1(Queue directQueue1, DirectExchange directExchange) {return BindingBuilder.bind(directQueue1).to(directExchange).with("black");}@Beanpublic Binding bindingDirectQueue2(Queue directQueue2, DirectExchange directExchange) {return BindingBuilder.bind(directQueue2).to(directExchange).with("green");}
}

2.DirectConfigListener.java 监听者

package com.sunxiansheng.consumer.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/*** Description: Direct交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Component
public class DirectConfigListener {@RabbitListener(queues = "direct.queue1")public void receive1(String message) {System.out.println("direct.queue1接收到的消息：" + message);}@RabbitListener(queues = "direct.queue2")public void receive2(String message) {System.out.println("direct.queue2接收到的消息：" + message);}
}

3.DirectConfig.java 生产者

package com.sunxiansheng.consumer.config;
import com.sunxiansheng.publisher.PublisherApplication;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
/*** Description: Direct交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@SpringBootTest(classes = PublisherApplication.class) // 指定启动类
public class DirectConfig {@Resourceprivate AmqpTemplate amqpTemplate;@Testpublic void send() {// 交换机String exchange = "direct.exchange";// 路由键String routingKey = "black";// 消息String message = "hello direct";// 发送消息amqpTemplate.convertAndSend(exchange, routingKey, message);}
}