热门标签
热门文章
- 1【深度学习】实验07 使用TensorFlow完成逻辑回归_tensorflow 逻辑回归
- 2迁移到 OpenAI Python API 库 1.x_openai migrate
- 3linux磁盘IO_linux 磁盘io
- 4springboot整合elasticsearch5.x以及IK分词器做全文检索_springboot elasticsearch 使用ik analyzer查询
- 5Win11配置VPN：L2TP连接尝试失败，因为安全层在初始化与远程计算机的协商时遇到了一个处理错误_win 11 l2tp无法连接
- 6mmcv及mmcv-full安装_mmcv1.5.0安装
- 7Mamba解读（FlashAttention,SSM,LSSL,S4,S5,Mamba）_mamba算法
- 8揭秘艺术的未来：AI绘画自动生成器的魔法
- 9【大数据工具】Spark 伪分布式、分布式集群搭建_spark伪分布式集群搭建
- 10由百鲤跃川主办首届中国企业采购服务产业数字大会于深圳圆满收官
当前位置: article > 正文
rabbitmq 学习(五) spring-boot-rabbit同时连接多个rabbitmq服务器收发消息_springboot 利用rabbitmq 与多个设备进行通信
作者：小丑西瓜9 | 2024-05-24 10:20:59
赞
踩
springboot 利用rabbitmq 与多个设备进行通信
上一遍博客介绍了使用spring-boot连接单个rabbitmq服务器发送和接收消息，但是在实际的项目中我们可能需要同时连接多个不同的rabbitmq服务器发送和接收消息。今天简单介绍下如何使用spring-boot连个多个rabbitmq服务器发送和接收消息。
1.配置文件中配置2个rabbitmq连接信息(这里我在2台虚拟机上安装rabbitmq)
server:
port: 8091
spring:
application:
name: rabbit-spring-boot
rabbitmq:
dev:
host: 192.168.98.22
port: 5672
username: root
password: root
test:
host: 192.168.98.23
port: 5672
username: root
password: root
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
2.创建rabbitmq连接配置类
@Configuration
public class RabbitMqConfig {
// 注意这里使用了primary注解
// 声明连接工厂连接开发服务器
@Primary
@Bean(name = "devConnectionFactory")
public ConnectionFactory devConnectionFactory(@Value("${spring.rabbitmq.dev.host}") String host,
@Value("${spring.rabbitmq.dev.port}") int port,
@Value("${spring.rabbitmq.dev.username}") String username,
@Value("${spring.rabbitmq.dev.password}") String password) {
// 使用@Value直接读取配置文件中的信息
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost(host);
connectionFactory.setPort(port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true);
connectionFactory.setPublisherReturns(true);
return connectionFactory;
}
// 声明连接工厂连接测试服务器
@Bean(name = "testConnectionFactory")
public ConnectionFactory testConnectionFactory(@Value("${spring.rabbitmq.test.host}") String host,
@Value("${spring.rabbitmq.test.port}") int port,
@Value("${spring.rabbitmq.test.username}") String username,
@Value("${spring.rabbitmq.test.password}") String password) {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost(host);
connectionFactory.setPort(port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true);
connectionFactory.setPublisherReturns(true);
return connectionFactory;
}
// 声明开发服务器rabbitTemplate
@Bean(name = "devRabbitTemplate")
public RabbitTemplate devRabbitTemplate(@Qualifier("devConnectionFactory") ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate=new RabbitTemplate(connectionFactory);
return rabbitTemplate;
}
// 声明测试服务器连接 rabbitTemplate
@Bean(name = "testRabbitTemplate")
public RabbitTemplate testRabbitTemplate(@Qualifier("testConnectionFactory") ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate=new RabbitTemplate(connectionFactory);
return rabbitTemplate;
}
/**
* 声明dev containerFactory
* @param rabbitListenerContainerFactoryConfigurer
* @param connectionFactory
* @return
*/
@Bean(name = "devContainerFactory")
public SimpleRabbitListenerContainerFactory devSimpleRabbitListenerContainerFactory(
SimpleRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer,
@Qualifier("devConnectionFactory") ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory containerFactory=new SimpleRabbitListenerContainerFactory();
rabbitListenerContainerFactoryConfigurer.configure(containerFactory,connectionFactory);
return containerFactory;
}
/**
* 声明 test containerFactory
* @param rabbitListenerContainerFactoryConfigurer
* @param connectionFactory
* @return
*/
@Bean(name = "testContainerFactory")
public SimpleRabbitListenerContainerFactory testSimpleRabbitListenerContainerFactory(
SimpleRabbitListenerContainerFactoryConfigurer rabbitListenerContainerFactoryConfigurer,
@Qualifier("testConnectionFactory") ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory containerFactory=new SimpleRabbitListenerContainerFactory();
rabbitListenerContainerFactoryConfigurer.configure(containerFactory,connectionFactory);
return containerFactory;
}
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
3.创建消息生产者
@Component
public class RabbitMQProducer {
// 指定dev服务器rabbitTemplate
@Resource(name = "devRabbitTemplate")
AmqpTemplate devAmqpTemplate;
// 指定test服务器rabbitTemplate
@Resource(name = "testRabbitTemplate")
AmqpTemplate testAmqpTemplate;
@Value("${open-care.rabbitmq.exchange.name}")
String exchangeName;
@Value("${open-care.rabbitmq.queue1-name}")
String queueName;
public void send(Object object) {
try {
Message message= MessageBuilder.withBody(JSON.toJSONString(object).getBytes("UTF-8"))
.setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setContentEncoding("UTF-8")
.setMessageId(UUID.randomUUID().toString()).build();
// 指定exchange交换器名称为fanout_exchange(fanout交换器没有路由键)
devAmqpTemplate.send("fanout_exchange","",message);
// 使用devAmqpTemplate将消息发送到开发服务器
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
public void sendMessage(Object object) {
try {
Message message= MessageBuilder.withBody(JSON.toJSONString(object).getBytes("UTF-8"))
.setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setContentEncoding("UTF-8")
.setMessageId(UUID.randomUUID().toString()).build();
// 指定交换器名称为fanout_exchange(fanout交换器没有路由键)
testAmqpTemplate.send("fanout_exchange","",message);
// 使用testAmqpTeplate将消息发送到测试服务器
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
4.创建消息消费者
@Component
public class RabbitMQConsumer {
@RabbitHandler
@RabbitListener(queues = "q.test1",containerFactory = "devContainerFactory")// 指定队列名称和容器工厂，接收开发服务器队列q.test1中的消息
public void receive(Object object) {
Message message=(Message)object;
byte bytes[]=null;
if (message != null) {
bytes=message.getBody();
}
try {
String msg=new String(bytes,"UTF-8");
System.out.println("192.168.98.22 q.test1------------------------"+msg);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@RabbitHandler
@RabbitListener(queues = "q.test2",containerFactory = "devContainerFactory")// 指定队列名称和容器工厂，接收开发服务器队列q.test2中的消息
public void receiveQueue1(Object object) {
Message message=(Message)object;
byte bytes[]=null;
if (message != null) {
bytes=message.getBody();
}
try {
String msg=new String(bytes,"UTF-8");
System.out.println("192.168.98.22 q.test2------------------------"+msg);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@RabbitHandler
@RabbitListener(queues = "q.test3",containerFactory = "devContainerFactory")// 指定队列名称和容器工厂，接收开发服务器队列q.test3中的消息
public void receiveQueue2(Object object) {
Message message=(Message)object;
byte bytes[]=null;
if (message != null) {
bytes=message.getBody();
}
try {
String msg=new String(bytes,"UTF-8");
System.out.println("192.168.98.22 q.test3------------------------"+msg);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@RabbitHandler
@RabbitListener(queues = "q.test1",containerFactory = "testContainerFactory")
// 指定队列名称和容器工厂，接收测试服务器队列q.test1中的消息
public void receive23message(Object object) {
Message message=(Message)object;
byte bytes[]=null;
if (message != null) {
bytes=message.getBody();
}
try {
String msg=new String(bytes,"UTF-8");
System.out.println("192.168.98.23 q.test1------------------------"+msg);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@RabbitHandler
@RabbitListener(queues = "q.fanout1",containerFactory = "testContainerFactory")// 指定队列名称和容器工厂，接收测试服务器队列q.fanout1中的消息
public void receiveFanoutQueue(Object object) {
Message message=(Message)object;
byte bytes[]=null;
if (message != null) {
bytes=message.getBody();
}
try {
String msg=new String(bytes,"UTF-8");
System.out.println("192.168.98.23 q.fanout1------------------------"+msg);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小丑西瓜9/article/detail/617000
推荐阅读
相关标签
