高级消息队列协议,RabbitMQ使用的是AMQP协议

时间:2019-10-06 23:58来源:美高梅手机游戏网站
/** * topic队列消费者 * * @author onlinever * @date 2018/8/17 */public interface TopicConsumer { /** * 消费的队列 * * @return 队列 */ RabbitQueueEnum getQueueEnum(); /** * 具体消费者的实现 * * @param message 消息 */ vo
/** * topic队列消费者 * * @author onlinever * @date 2018/8/17 */public interface TopicConsumer { /** * 消费的队列 * * @return 队列 */ RabbitQueueEnum getQueueEnum(); /** * 具体消费者的实现 * * @param message 消息 */ void handleMessage(String message);}

RabbitMQ消费失败的处理

RabbitMQ采用消息应答机制,即消费者收到一个消息之后,需要发送一个应答,然后RabbitMQ才会将这个消息从队列中删除,如果消费者在消费过程中出现异常,断开连接切没有发送应答,那么RabbitMQ会将这个消息重新投递。

修改一下消费者的代码:

//接收到消息事件
consumer.Received += (ch, ea) =>
{
    var message = Encoding.UTF8.GetString(ea.Body);

    Console.WriteLine($"收到消息: {message}");

    Console.WriteLine($"收到该消息[{ea.DeliveryTag}] 延迟10s发送回执");
    Thread.Sleep(10000);
    //确认该消息已被消费
    channel.BasicAck(ea.DeliveryTag, false);
    Console.WriteLine($"已发送回执[{ea.DeliveryTag}]");
};

演示:

图片 1

从图中可以看出,设置了消息应答延迟10s,如果在这10s中,该消费者断开了连接,那么消息会被RabbitMQ重新投递。

RabbitMQ 的 Hello Demo

安装就不说了,建议按照官方文档上做。先贴代码,稍后解释,代码如下:

配置 交换机,队列,交换机与队列的绑定,消息监视容器:

@Configuration
@Data
public class RabbitMQConfig {

    final static String queueName = "spring-boot";

    @Bean
    Queue queue() {
        return new Queue(queueName, false);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("spring-boot-exchange");
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(queueName);
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(listenerAdapter);
        return container;
    }

    @Bean
    Receiver receiver() {
        return new Receiver();
    }
    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }
}

配置接收信息者(即消费者):

public class Receiver {

    private CountDownLatch latch = new CountDownLatch(1);

    public void receiveMessage(String message) {
        System.out.println("Received <" + message + ">");
        latch.countDown();
    }

    public CountDownLatch getLatch() {
        return latch;
    }
}

配置发送信息者(即生产者):

@RestController
public class Test {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @RequestMapping(value = "/test/{abc}",method = RequestMethod.GET)
    public String test(@PathVariable(value = "abc") String abc){
        rabbitTemplate.convertAndSend("spring-boot", abc + " from RabbitMQ!");
        return  "abc";
    }
}

以上便可实现一个简单的 RabbitMQ Demo,具体代码在:点这里

那么,这里,分为三个部分分析:发消息,交换机队列,收消息。

  • 对于发送消息:我们一般可以使用 RabbitTemplate,这个是 Spring 封装给了我们,便于我们发送信息,我们调用 rabbitTemplate.convertAndSend("spring-boot", xxx); 即可发送信息。
  • 对于交换机队列:如上代码,我们需要配置交换机 TopicExchange,配置队列 Queue,并且配置他们之间的绑定 Binding
  • 对于接受消息:首先需要创建一个消息监听容器,然后把我们的接受者注册到该容器中,这样,队列中有信息,那么就会调用接收者的对应的方法。如上代码 container.setMessageListener(listenerAdapter); 其中,MessageListenerAdapter 可以看做是 我们接收者的一个包装类,new MessageListenerAdapter(receiver, "receiveMessage"); 指明了如果有消息来,那么调用接收者哪个方法进行处理。
  1. 其他消费者使用@RabbitListener方式
定义生产者
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory
{
    UserName = "admin",//用户名
    Password = "admin",//密码
    HostName = "192.168.157.130"//rabbitmq ip
};

//创建连接
var connection = factory.CreateConnection();
//创建通道
var channel = connection.CreateModel();
//声明一个队列
channel.QueueDeclare("hello", false, false, false, null);

Console.WriteLine("nRabbitMQ连接成功,请输入消息,输入exit退出!");

string input;
do
{
    input = Console.ReadLine();

    var sendBytes = Encoding.UTF8.GetBytes(input);
    //发布消息
    channel.BasicPublish("", "hello", null, sendBytes);

} while (input.Trim().ToLower()!="exit");
channel.Close();
connection.Close();

RabbitMQ 在生产环境下运用和出现的问题

在生产环境中,由于 Spring 对 RabbitMQ 提供了一些方便的注解,所以首先可以使用这些注解。例如:

  • @EnableRabbit:@EnableRabbit 和 @Configuration 注解在一个类中结合使用,如果该类能够返回一个 RabbitListenerContainerFactory 类型的 bean,那么就相当于能够把该终端(消费端)和 RabbitMQ 进行连接。Ps:(生成端不是通过 RabbitListenerContainerFactory 来和 RabbitMQ 连接,而是通过 RabbitTemplate )
  • @RabbitListener:当对应的队列中有消息的时候,该注解修饰下的方法会被执行。
  • @RabbitHandler:接收者可以监听多个队列,不同的队列消息的类型可能不同,该注解可以使得不同的消息让不同方法来响应。

具体这些注解的使用,可以参考这里的代码:点这里

首先,生产环境下的 RabbitMQ 可能不会在生产者或者消费者本机上,所以需要重新定义 ConnectionFactory,即:

@Bean
ConnectionFactory connectionFactory() {
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
    connectionFactory.setUsername(userName);
    connectionFactory.setPassword(password);
    connectionFactory.setVirtualHost(vhost);
    return connectionFactory;
}

这里,可以重新设置需要连接的 RabbitMQ 的 ip,端口,虚拟主机,用户名,密码。

然后,可以先从生产端考虑,生产端需要连接 RabbitMQ,那么可以通过 RabbitTemplate 进行连接。 Ps:(RabbitTemplate 用于生产端发送消息到交换机中),如下代码:

@Bean(name="myTemplate")
RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMessageConverter(integrationEventMessageConverter());
    template.setExchange(exchangeName);
    return template;
}

在该代码中,new RabbitTemplate(connectionFactory); 设置了生产端连接到RabbitMQ,template.setMessageConverter(integrationEventMessageConverter()); 设置了 生产端发送给交换机的消息是以什么格式的,在 integrationEventMessageConverter() 代码中:

public MessageConverter integrationEventMessageConverter() {
    Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();
    return messageConverter;
}

如上 Jackson2JsonMessageConverter 指明了 JSON。上述代码的最后 template.setExchange(exchangeName); 指明了 要把生产者要把消息发送到哪个交换机上。

有了上述,那么,我们即可使用 rabbitTemplate.convertAndSend("spring-boot", xxx); 发送消息,xxx 表示任意类型,因为上述的设置会帮我们把这些类型转化成 JSON 传输。

接着,生产端发送我们说过了,那么现在可以看看消费端:

对于消费端,我们可以只创建 SimpleRabbitListenerContainerFactory,它能够帮我们生成 RabbitListenerContainer,然后我们再使用 @RabbitListener 指定接收者收到信息时处理的方法。

@Bean(name="myListenContainer")
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setMessageConverter(integrationEventMessageConverter());
    factory.setConnectionFactory(connectionFactory());
    return factory;
}

这其中 factory.setMessageConverter(integrationEventMessageConverter()); 指定了我们接受消息的时候,以 JSON 传输的消息可以转换成对应的类型传入到方法中。例如:

@Slf4j
@Component
@RabbitListener(containerFactory = "helloRabbitListenerContainer",queues = "spring-boot")
public class Receiver {
    @RabbitHandler
    public void receiveTeacher(Teacher teacher) {
        log.info("##### = {}",teacher);
    }
}

可能出现的问题:

2.3 队列枚举

运行

图片 2

启动了一个生产者,两个消费者,可以看见两个消费者都能收到消息,消息投递到哪个消费者是由RabbitMQ决定的。

注:这份文档是我和几个朋友学习后一起完成的。

/** * @author onlinever * @date 2018/09/06 */public enum RabbitExchangeEnum { /** * rabbit交换机名称 * 默认一个应用设置一个交换机 * exchange.{0}.{1} * 0: 交换机类型 direct、topic、fanout、headers * 1: 应用名称 */ DIRECT_EXCHANGE("directExchange", "exchange.direct.gateway", RabbitExchangeTypeEnum.NORMAL_QUEUE), FANOUT_EXCHANGE("fanoutExchange", "exchange.fanout.gateway", RabbitExchangeTypeEnum.FANOUT_QUEUE), TOPIC_EXCHANGE("topicExchange", "exchange.topic.gateway", RabbitExchangeTypeEnum.TOPIC_QUEUE),; /** * 交换机beanName */ private String beanName; /** * 交换机key */ private String exchangeName; /** * 交换机类型 */ private RabbitExchangeTypeEnum rabbitExchangeTypeEnum; RabbitExchangeEnum(String beanName, String exchangeName, RabbitExchangeTypeEnum rabbitExchangeTypeEnum) { this.beanName = beanName; this.exchangeName = exchangeName; this.rabbitExchangeTypeEnum = rabbitExchangeTypeEnum; } public String getExchangeName() { return exchangeName; } public String getBeanName() { return beanName; } public RabbitExchangeTypeEnum getRabbitExchangeTypeEnum() { return rabbitExchangeTypeEnum; }}

RabbitMQ安装

RabbitMQ安装,网上已经有许多教程了,这里简单介绍一下在CentOS下安装RabbitMQ。使用的版本为3.6.12最新版。

1.首先安装erlang

rpm -Uvh https://www.rabbitmq.com/releases/erlang/erlang-19.0.4-1.el7.centos.x86_64.rpm

2.然后安装socat

yun install socat

3.最后安装RabbitMQ

rpm -Uvh https://www.rabbitmq.com/releases/rabbitmq-server/v3.6.12/rabbitmq-server-3.6.12-1.el7.noarch.rpm

binding?

exchange和queue通过routing-key关联,这两者之间的关系是就是binding。如下图所示,X表示交换机,红色表示队列,交换机通过一个routing-key去binding一个queue,routing-key有什么作用呢?看Direct exchange类型交换机。

图片 3

实现Direct,Fanout,Topic和死信转发方式实现的延迟队列

.NET Core 使用RabbitMQ

消息持久化

在生产环境中,我们需要考虑万一生产者挂了,消费者挂了,或者 rabbitmq 挂了怎么样。一般来说,如果生产者挂了或者消费者挂了,其实是没有影响,因为消息就在队列里面。那么万一 rabbitmq 挂了,之前在队列里面的消息怎么办,其实可以做消息持久化,RabbitMQ 会把信息保存在磁盘上。

做法是可以先从 Connection 对象中拿到一个 Channel 信道对象,然后再可以通过该对象设置 消息持久化。

使用RabbitMQ的Exchange

前面我们可以看到生产者将消息投递到Queue中,实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是,生产者将消息发送到Exchange(交换器),由Exchange将消息路由到一个或多个Queue中(或者丢弃)

图片 4

AMQP协议中的核心思想就是生产者和消费者隔离,生产者从不直接将消息发送给队列。生产者通常不知道是否一个消息会被发送到队列中,只是将消息发送到一个交换机。先由Exchange来接收,然后Exchange按照特定的策略转发到Queue进行存储。同理,消费者也是如此。Exchange 就类似于一个交换机,转发各个消息分发到相应的队列中。

RabbitMQ提供了四种Exchange模式:direct,fanout,topic,header 。但是 header模式在实际使用中较少,所以这里只介绍前三种模式。

Exchange不是消费者关心的,所以消费者的代码完全不用变,用上面的消费者就行了。
由于避免文章过长,影响阅读,所以只贴了部分代码,但是demo里面是完整可运行的,详细代码请查看demo。

目录

  • RabbitMQ 概念
  • exchange交换机机制
    • 什么是交换机
    • binding?
    • Direct Exchange交换机
    • Topic Exchange交换机
    • Fanout Exchange交换机
    • Header Exchange交换机
  • RabbitMQ 的 Hello - Demo(springboot实现)
  • RabbitMQ 的 Hello Demo(spring xml实现)
  • RabbitMQ 在生产环境下运用和出现的问题
    • Spring RabbitMQ 注解
    • 消息的 JSON 传输
    • 消息持久化,断线重连,ACK。

RabbitMQ简介

AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然。

AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。
RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。

RabbitMQ提供了可靠的消息机制、跟踪机制和灵活的消息路由,支持消息集群和分布式部署。适用于排队算法、秒杀活动、消息分发、异步处理、数据同步、处理耗时任务、CQRS等应用场景。

Fanout Exchange

扇形交换机,该交换机会把消息发送到所有binding到该交换机上的queue。这种是publisher/subcribe模式。用来做广播最好。
所有该exchagne上指定的routing-key都会被ignore掉。

The fanout copies and routes a received message to all queues that are bound to it regardless of routing keys or pattern matching as with direct and topic exchanges. Keys provided will simply be ignored.

编辑:美高梅手机游戏网站 本文来源:高级消息队列协议,RabbitMQ使用的是AMQP协议

关键词: