RabbitMQ

RabbitMQ是实现AMQP(高级消息队列协议)消息中间件的一种, kafuka是另外一种
RabbitMQ主要是为了实现系统之间的双向解耦而实现的,消息的发送者无需知道消息使用者的存在，反之亦然
当生产者大量产生数据时, 消费者无法快速消费, 那么需要一个中间层,保存这个数据
AMQP的主要特征是面向消息(Message)、队列(Queue)、路由(Exchange包括点对点和发布/订阅)、可靠性、安全

1.环境搭建

使用docker启动已经做好的镜像即可
docker pull rabbitmq:3-management
docker run -itd --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

2.依赖

pom.xml 增加依赖

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

3.配置

在需要使用rabbitmq的项目配置文件中增加 rabbitmq 相关配置项

rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    listener: 
      direct: 
        acknowledge-mode: manual 
      simple: 
        acknowledge-mode: manual

4.相关概念

一般来说消息队列有三个概念：发消息者、队列、收消息者，
RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和队列之间, 加入了交换器 (Exchange).
这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

交换机(Exchange)
交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。
交换机有四种类型：Direct, topic, Headers and Fanout
Direct：direct 类型的行为是“先匹配, 再投送”. 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
Topic：按规则转发消息（最灵活）
Headers：设置header attribute参数类型的交换机
Fanout：转发消息到所有绑定队列

项目中使用的是Topic模式

topicExchange.jpg

消费者模糊匹配路由Key，进行消息消费，路由Key必须是一串字符，用句号（.）隔开，比如说，applet.create，applet.#
主要有两种模糊匹配：# 匹配一个或多个，* 匹配一个，一般使用#号匹配多个，*号用的比较少。

如果消费者和生产者在分属两个不同服务进程中，建议创建队列的操作放在消费者那个进程中做，因为消费者启动的时候需要监听队列，若放在了生产者
进程中创建而生产者又没有启动，则消费者启动的时候会报错。交换机在生产者和消费者都可以创建，就看谁先启动谁先创建了

5.代码

生产者进程相关代码
rabbitmq使用前的准备配置代码

@Configuration
public class RabbitmqConfig {
    //使用Jackson2JsonMessageConverter 消息传递后转对象 
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    //创建一个名为applet.exchange的交换机
    @Bean
    TopicExchange appletExchange() {
        return new TopicExchange("applet.exchange", true, false);
    }
}

发送消息到交换机代码

@Service
public class UpdaterServiceImpl implements UpdaterService {

    @Autowired
    RabbitTemplate rabbitTemplate;

    public void create(AppletRequestDTO dto) {
        rabbitTemplate.convertAndSend("applet.exchange", "applet.create", dto);
    }

    public void update(AppletRequestDTO dto) {
        rabbitTemplate.convertAndSend("applet.exchange", "applet.update", dto);
    }
}

消费者进程相关代码
rabbitmq使用前的准备配置代码

@Configuration
public class RabbitmqConfig {
    //使用Jackson2JsonMessageConverter 消息传递后转对象 
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    //创建一个名为applet.exchange的交换机
    @Bean
    TopicExchange appletExchange() {
        return new TopicExchange("applet.exchange", true, false);
    }
    //创建一个名为search_index_queue队列
    @Bean
    Queue searchIndexQueue() {
        return new Queue("search_index_queue", true);
    }
    //创建一个名为search_update_queue队列
    @Bean
    Queue searchUpdateQueue() {
        return new Queue("search_update_queue", true);
    }

    //队列绑定到路由器上并指定使用什么路由Key
    @Bean
    Binding searchIndexBinding() {
        return BindingBuilder.bind(searchIndexQueue()).to(appletExchange()).with("applet.create");
    }

    @Bean
    Binding searchUpdateBinding() {
        return BindingBuilder.bind(searchUpdateQueue()).to(appletExchange()).with("applet.update");
    }
}

消费监听和接收消息队列的消息代码
有2种方式配置RabbitListener，一种使用RabbitListener标注类且使用RabbitHandler标注方法，另一种使用RabbitListener标注方法

@Component
@Slf4j
public class RabbitmqListener {

    @Autowired
    AppletSearchService appletSearchService;

    @RabbitListener(queues = "search_index_queue")
    public void handlerIndex(AppletRequestDTO dto, Channel channel, Message message) {
        try {
            //处理消息
            appletSearchService.index(dto);
            //通知消息处理完毕
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            e.printStackTrace();
            //丢弃这条消息
            log.error("索引消息处理错误. msg:{}", e.getMessage());
        }
    }

    @RabbitListener(queues = "search_update_queue")
    public void handlerUpdate(AppletRequestDTO dto, Channel channel, Message message) {
        try {
            appletSearchService.update(dto);
            //通知消息处理完毕
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
            //丢弃这条消息
            //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            log.error("更新文档消息消息处理错误. msg:{}", e.getMessage());
        }
    }
}