标签：activemq 云服务 converter ide lang except tee getname 功能
在大多数应用中，可以通过消息服务中间件来提升系统的异步通信、扩展解耦和流量削峰等能力。
当消息发送者发送消息后，将由消息代理接管，消息代理保证消息传递到指定目的地。
AMQP（Advanced Message Queuing Protocol）：高级消息队列协议，也是一个消息代理的规范，兼容 JMS，RabbitMQ是 AMQP 的实现。
简介：RabbitMQ 是一个由 erlang 开发的 AMQP 的开源实现。
核心概念：
Exchange 类型：Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers（性能较差，几乎不用）。
拉取镜像（阿里云服务器上）：
[root@izwz9d74k4cznxtxjeeur9z ~]# docker pull rabbitmq:3-management镜像实例化为容器：
[root@izwz9d74k4cznxtxjeeur9z ~]# docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq a7f2574d507f
# 第一个端口映射为 rabbitmq 客户端服务，第二个端口映射为 web 管理界面测试访问（IP + 端口）：
输入账号：guest 密码：guest
以上图中的 3 种交换器、4 个队列为例，测试 RabbitMQ：
添加一个名为 exchange.direct ，类型为 direct 交换器；添加一个名为 exchange.fanout，类型为 fanout 的交换器；添加一个名为 exchange.topic，类型为 topic 的交换器。
添加 4 个队列，分别名为 atguigu、atguigu.news、atguigu.emps 和 gulixueyuan.news。
添加交换器 和 队列之间的绑定（Routing Key）：①为 exchange.direct 与 4 个队列之间添加 Bindings，Routing Key 分别为队列名 atguigu、atguigu.news、atguigu.emps 和 gulixueyuan.news。（direct 类型交换器精确匹配路由键）
②为 exchange.fanout 与 4 个队列之间添加 Bindings，Routing Key 分别为队列名 atguigu、atguigu.news、atguigu.emps 和 gulixueyuan.news。（fanout 类型交换器与路由键无关，它是广播的）
③为 exchange.topic 与 4 个队列之间添加如示意图所示的 Bindings。绑定结果如下：
4.发布消息：①向 exchange.direct 发送 4 条消息，消息的 Routing key 分别为 atguigu、atguigu.news、atguigu.emps 和 gulixueyuan.news。
预测 4 个队列各自有一条消息，消息发送到队列根据路由键来的。
依次读取消息并从队列中移除：
? ②向 exchange.fanout 中 发送一条消息，路由键随意，由于它是广播的，所以 4 个队列都会得到该消息。
?
?
? 移除所有消息。
? ③向 exchan.topic 中 发送一条消息，路由键为 atguigu.news,由于队列 atguigu 绑定的路由键为 atguigu.#,匹配成功，收到消息，队列 gulixueyuan.news 绑定的路由键为*.news，匹配成功收到消息。同理另外 2 个队列也会收到消息。若发送的消息键值为 abc.news, 则只有队列 gulixueyuan.news 和 guigu.news 收到消息。
RabbitAutoConfiguration 自动配置了连接工厂 rabbitConnectionFactory
@Configuration
@ConditionalOnClass({ RabbitTemplate.class, Channel.class })
@EnableConfigurationProperties(RabbitProperties.class)
@Import(RabbitAnnotationDrivenConfiguration.class)
public class RabbitAutoConfiguration {
@Configuration
@ConditionalOnMissingBean(ConnectionFactory.class)
protected static class RabbitConnectionFactoryCreator {
@Bean
public CachingConnectionFactory rabbitConnectionFactory(
RabbitProperties properties,
ObjectProvider<ConnectionNameStrategy> connectionNameStrategy)
throws Exception {
PropertyMapper map = PropertyMapper.get();
CachingConnectionFactory factory = new CachingConnectionFactory(
getRabbitConnectionFactoryBean(properties).getObject());
map.from(properties::determineAddresses).to(factory::setAddresses);
map.from(properties::isPublisherConfirms).to(factory::setPublisherConfirms);RabbitProperties 封装了 RabbitMQ 的配置
@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitProperties {
/**
* RabbitMQ host.
*/
private String host = "localhost";
/**
* RabbitMQ port.
*/
private int port = 5672;
/**
* Login user to authenticate to the broker.
*/
private String username = "guest";AmqpAdmin ： RabbitMQ 系统管理功能组件，创建和删除 Queue，Exchange，Binding
@Bean
@ConditionalOnSingleCandidate(ConnectionFactory.class)
@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "dynamic", matchIfMissing = true)
@ConditionalOnMissingBean
public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
}@ManagedResource(description = "Admin Tasks")
public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, ApplicationEventPublisherAware,
BeanNameAware, InitializingBean {
...
@Override
public void declareExchange(final Exchange exchange) {
try {
this.rabbitTemplate.execute(channel -> {
declareExchanges(channel, exchange);
return null;
});
}
catch (AmqpException e) {
logOrRethrowDeclarationException(exchange, "exchange", e);
}
}
@Override
@ManagedOperation(description = "Delete an exchange from the broker")
public boolean deleteExchange(final String exchangeName) {
return this.rabbitTemplate.execute(channel -> { // NOSONAR never returns nullRabbitTemplate ：给 RabbitMQ 发送和接受消息
@Bean
@ConditionalOnSingleCandidate(ConnectionFactory.class)
@ConditionalOnMissingBean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
PropertyMapper map = PropertyMapper.get();
RabbitTemplate template = new RabbitTemplate(connectionFactory);
MessageConverter messageConverter = this.messageConverter.getIfUnique();
if (messageConverter != null) {
template.setMessageConverter(messageConverter);
}public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count/comment density
implements BeanFactoryAware, RabbitOperations, MessageListener,
ListenerContainerAware, PublisherCallbackChannel.Listener, Lifecycle, BeanNameAware {
...
@Override
public void convertAndSend(Object object) throws AmqpException {
convertAndSend(this.exchange, this.routingKey, object, (CorrelationData) null);
}
application.properties：
spring.rabbitmq.host=xx.xx.xx.xx
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest使用 AmqpAdmin 创建 Exchange、队列和相应的 Bindings。
package com.yunche;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringRabbitmqApplicationTests {
@Autowired
AmqpAdmin amqpAdmin;
@Test
public void contextLoads() {
amqpAdmin.declareExchange(new DirectExchange("test-direct-exchange"));
System.out.println("创建了一个 direct 类型的 exchange");
amqpAdmin.declareQueue(new Queue("test-queue"));
System.out.println("创建了一个队列");
amqpAdmin.declareBinding(new Binding("test-queue", Binding.DestinationType.QUEUE, "test-direct-exchange", "test-routingkey", null));
System.out.println("添加了绑定，路由键为 test-routingkey");
}
}使用 RabbitTemplate 发送和接收消息：
@Autowired
RabbitTemplate rabbitTemplate;
@Test
public void sendMsg() {
rabbitTemplate.convertAndSend("test-direct-exchange","test-routingkey", "hello world");
}@Test
public void receiveMsg() {
String msg = (String)rabbitTemplate.receiveAndConvert("test-queue");
System.out.println(msg);
} /*Output:hello world*/@Test
public void sendObject() {
rabbitTemplate.convertAndSend("test-direct-exchange","test-routingkey", new Book("红楼梦", "曹雪芹"));
}
private static class Book implements Serializable {
private String name;
private String author;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Book(String name, String author) {
this.name = name;
this.author = author;
}
public String getAuthor() {
return author;
}
public void setAuthor(String author) {
this.author = author;
}
}
@Test
public void receiveObject() {
Book b = (Book)rabbitTemplate.receiveAndConvert("test-queue");
System.out.println(b.name);
} /*Output: 红楼梦*/@RabbitListener 监听消息队列的内容；首先用@EnableRabbit 开启基于注解的 RabbitMQ 模式
package com.yunche;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@EnableRabbit
@SpringBootApplication
public class SpringRabbitmqApplication {
public static void main(String[] args) {
SpringApplication.run(SpringRabbitmqApplication.class, args);
}
}首先运行单元测试中的 sendObject() 方法发送消息，然后控制台立即输出消息。
@RabbitListener(queues = "test-queue")
public void msgListener(Book book) {
System.out.println(book.getAuthor());
} /*Output：曹雪芹*/尚硅谷.Spring Boot 高级
标签：activemq 云服务 converter ide lang except tee getname 功能
原文地址：https://www.cnblogs.com/yunche/p/10351207.html
