springboot 集成rabbitmq 并采用ack模式以

作者: 路人甲java | 来源:发表于2020-06-17 14:11 被阅读0次

rabbitmq简介

rabbitmq 是spring所在公司Pivotal自己的产品是基于AMQP高级队列协议的消息中间件采用erlang开发因此安装需要erlang环境具体安装根据自己的环境因为跟spring有共同的血缘关系所以spring 全家桶对其的支持应该是相当完善的

** 简单概念**

一般消息队列都是生产者将消息发送到队列消费者监听队列进行消费 rabbitmq 一个虚拟主机（默认 /）持有一个或者多个交换机（Exchange）用户只能在虚拟主机的粒度进行权限控制交换机根据一定的策略（RoutingKey）绑定(Binding)到队列(Queue)上这样生产者和队列就没有直接联系而是将消息发送的交换机交换机再把消息转发到对应绑定的队列上此处需要详细熟悉rabbitmq的工作流程不清楚可以找相关资料进行学习

上面说了 Exchange 作为rabbitmq的一个独特的重要的概念这里有必要着重强调一下我们从 spring对rabbitmq的封装来解读一下这个东西

package org.springframework.amqp.core;

/**
 * Constants for the standard Exchange type names.
 *
 * @author Mark Fisher
 * @author Gary Russell
 */
public abstract class ExchangeTypes {

    public static final String DIRECT = "direct";

    public static final String TOPIC = "topic";

    public static final String FANOUT = "fanout";

    public static final String HEADERS = "headers";

    public static final String SYSTEM = "system";

    /**
     * The constant to represent {@code x-delayed-message} exchange mode.
     * @deprecated since 1.6.4, it's not a user-available exchange type,
     * the delayed {@code boolean} is used for that.
     */
    @Deprecated
    public static final String DELAYED = "x-delayed-message";
}

上面是交换机类型的定义类说明了6种交换机类型最后一种因为即将弃用所以是五种我们常用的有四种下面这个建造类说明了一切

package org.springframework.amqp.core;

import java.util.Map;

/**
 * Builder providing a fluent API for building {@link Exchange}s.
 *
 * @author Gary Russell
 * @since 1.6
 *
 */
public final class ExchangeBuilder extends AbstractBuilder {

    private final String name;

    private final String type;

    private boolean durable;

    private boolean autoDelete;

    private boolean internal;

    private boolean delayed;

    /**
     * Construct an instance of the appropriate type.
     * @param name the exchange name
     * @param type the type name
     * @see ExchangeTypes
     * @since 1.6.7
     */
    public ExchangeBuilder(String name, String type) {
        this.name = name;
        this.type = type;
    }

    /**
     * Return a {@link DirectExchange} builder.
     * @param name the name.
     * @return the builder.
     */
    public static ExchangeBuilder directExchange(String name) {
        return new ExchangeBuilder(name, ExchangeTypes.DIRECT);
    }

    /**
     * Return a {@link TopicExchange} builder.
     * @param name the name.
     * @return the builder.
     */
    public static ExchangeBuilder topicExchange(String name) {
        return new ExchangeBuilder(name, ExchangeTypes.TOPIC);
    }

    /**
     * Return a {@link FanoutExchange} builder.
     * @param name the name.
     * @return the builder.
     */
    public static ExchangeBuilder fanoutExchange(String name) {
        return new ExchangeBuilder(name, ExchangeTypes.FANOUT);
    }

    /**
     * Return a {@link HeadersExchange} builder.
     * @param name the name.
     * @return the builder.
     */
    public static ExchangeBuilder headersExchange(String name) {
        return new ExchangeBuilder(name, ExchangeTypes.HEADERS);
    }

    /**
     * Set the auto delete flag.
     * @return the builder.
     */
    public ExchangeBuilder autoDelete() {
        this.autoDelete = true;
        return this;
    }

    /**
     * Set the durable flag to true.
     * @return the builder.
     * @deprecated - in 2.0, durable will be true by default
     * @see #durable(boolean)
     */
    @Deprecated
    public ExchangeBuilder durable() {
        this.durable = true;
        return this;
    }

    /**
     * Set the durable flag.
     * @param durable the durable flag (default false).
     * @return the builder.
     * @see #durable
     */
    public ExchangeBuilder durable(boolean durable) {
        this.durable = durable;
        return this;
    }

    /**
     * Add an argument.
     * @param key the argument key.
     * @param value the argument value.
     * @return the builder.
     */
    public ExchangeBuilder withArgument(String key, Object value) {
        getOrCreateArguments().put(key, value);
        return this;
    }

    /**
     * Add the arguments.
     * @param arguments the arguments map.
     * @return the builder.
     */
    public ExchangeBuilder withArguments(Map<String, Object> arguments) {
        this.getOrCreateArguments().putAll(arguments);
        return this;
    }

    /**
     * Set the internal flag.
     * @return the builder.
     */
    public ExchangeBuilder internal() {
        this.internal = true;
        return this;
    }

    /**
     * Set the delayed flag.
     * @return the builder.
     */
    public ExchangeBuilder delayed() {
        this.delayed = true;
        return this;
    }

    public Exchange build() {
        AbstractExchange exchange;
        if (ExchangeTypes.DIRECT.equals(this.type)) {
            exchange = new DirectExchange(this.name, this.durable, this.autoDelete, getArguments());
        }
        else if (ExchangeTypes.TOPIC.equals(this.type)) {
            exchange = new TopicExchange(this.name, this.durable, this.autoDelete, getArguments());
        }
        else if (ExchangeTypes.FANOUT.equals(this.type)) {
            exchange = new FanoutExchange(this.name, this.durable, this.autoDelete, getArguments());
        }
        else if (ExchangeTypes.HEADERS.equals(this.type)) {
            exchange = new HeadersExchange(this.name, this.durable, this.autoDelete, getArguments());
        }
        else {
            throw new IllegalStateException("Invalid type: " + this.type);
        }
        exchange.setInternal(this.internal);
        exchange.setDelayed(this.delayed);
        return exchange;
    }

}

image

这四种的说明

Direct: 先策略匹配到对应绑定的队列后才会被投送到该队列交换机跟队列必须是精确的对应关系这种最为简单
Topic: 转发消息主要是根据通配符。在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息这种可以认为是Direct 的灵活版
Headers:也是根据规则匹配, 相较于 direct 和 topic 固定地使用 routingkey , headers 则是一个自定义匹配规则的类型
在队列与交换器绑定时会设定一组键值对规则消息中也包括一组键值对( headers 属性) 当这些键值对有一对或全部匹配时消息被投送到对应队列
Fanout : 消息广播模式不管路由键或者是路由模式 *会把消息发给绑定给它的全部队列 *如果配置了routingkey会被忽略

**springboot集成rabbitmq **

在熟悉了相关概念后我们开始搞一搞这个东西首先你要安装好rabbitmq 相关方法资料很多此处不表在本机安装好并启用了管理页面后打开 localhost:15672 会显示一个管理页面如下可以进行一些可视化操作

image

新建springboot工程 springboot 版本 1.5.10 依赖如下

    <dependencies>
        <!--amqp rabbitmq 依赖必须 必须-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!--springboot单元测试 选-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!--springboot健康监控 选-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!--web支持  选-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

application.yml 配置文件 rabbitmq 相关：

spring:
  rabbitmq:
    username: rabbitAdmin
    password: 123456789
#    支持发布确认
    publisher-confirms: true
#    支持发布返回
    publisher-returns: true
    listener:
      simple:
#      采用手动应答
        acknowledge-mode: manual
#        当前监听容器数
        concurrency: 1
#        最大数
        max-concurrency: 1
#        是否支持重试
        retry:
          enabled: true
#        日志配置 
logging:
  config: classpath:logback.xml

定制模版类声明交换机队列绑定交换机到队列

这里声明了Direct 交换机并通过路由键绑定到一个队列中来测试Direct模式

    声明了Fanout交换机  并绑定到2个队列    来测试广播模式

package cn.felord.message.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * 队列配置.
 *
 * @author dax.
 * @version v1.0
 * @since 2018 /2/23 14:28
 */
@Configuration
public class RabbitConfig {
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 定制化amqp模版      可根据需要定制多个
     * 
     * 
     * 此处为模版类定义 Jackson消息转换器
     * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调   即消息发送到exchange  ack
     * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调  即消息发送不到任何一个队列中  ack
     *
     * @return the amqp template
     */
//    @Primary
    @Bean
    public AmqpTemplate amqpTemplate() {
        Logger log = LoggerFactory.getLogger(RabbitTemplate.class);
//          使用jackson 消息转换器
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setEncoding("UTF-8");
//        开启returncallback     yml 需要 配置    publisher-returns: true
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            String correlationId = message.getMessageProperties().getCorrelationIdString();
            log.debug("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {}  路由键: {}", correlationId, replyCode, replyText, exchange, routingKey);
        });
        //        消息确认  yml 需要配置   publisher-returns: true
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.debug("消息发送到exchange成功,id: {}", correlationData.getId());
            } else {
                log.debug("消息发送到exchange失败,原因: {}", cause);
            }
        });
        return rabbitTemplate;
    }

    /* ----------------------------------------------------------------------------Direct exchange test--------------------------------------------------------------------------- */

    /**
     * 声明Direct交换机 支持持久化.
     *
     * @return the exchange
     */
    @Bean("directExchange")
    public Exchange directExchange() {
        return ExchangeBuilder.directExchange("DIRECT_EXCHANGE").durable(true).build();
    }

    /**
     * 声明一个队列 支持持久化.
     *
     * @return the queue
     */
    @Bean("directQueue")
    public Queue directQueue() {
        return QueueBuilder.durable("DIRECT_QUEUE").build();
    }

    /**
     * 通过绑定键 将指定队列绑定到一个指定的交换机 .
     *
     * @param queue    the queue
     * @param exchange the exchange
     * @return the binding
     */
    @Bean
    public Binding directBinding(@Qualifier("directQueue") Queue queue, @Qualifier("directExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("DIRECT_ROUTING_KEY").noargs();
    }

    /* ----------------------------------------------------------------------------Fanout exchange test--------------------------------------------------------------------------- */

    /**
     * 声明 fanout 交换机.
     *
     * @return the exchange
     */
    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange() {
        return (FanoutExchange) ExchangeBuilder.fanoutExchange("FANOUT_EXCHANGE").durable(true).build();
    }

    /**
     * Fanout queue A.
     *
     * @return the queue
     */
    @Bean("fanoutQueueA")
    public Queue fanoutQueueA() {
        return QueueBuilder.durable("FANOUT_QUEUE_A").build();
    }

    /**
     * Fanout queue B .
     *
     * @return the queue
     */
    @Bean("fanoutQueueB")
    public Queue fanoutQueueB() {
        return QueueBuilder.durable("FANOUT_QUEUE_B").build();
    }

    /**
     * 绑定队列A 到Fanout 交换机.
     *
     * @param queue          the queue
     * @param fanoutExchange the fanout exchange
     * @return the binding
     */
    @Bean
    public Binding bindingA(@Qualifier("fanoutQueueA") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    /**
     * 绑定队列B 到Fanout 交换机.
     *
     * @param queue          the queue
     * @param fanoutExchange the fanout exchange
     * @return the binding
     */
    @Bean
    public Binding bindingB(@Qualifier("fanoutQueueB") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }
}

编写监听器来监听队列消息

package cn.felord.message.comsumer;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 监听器.
 *
 * @author dax.
 * @version v1.0
 * @since 2018 /2/24 9:36
 */
@Component
public class Receiver {
    private static final Logger log= LoggerFactory.getLogger(Receiver.class);
    /**
     * FANOUT广播队列监听一.
     *
     * @param message the message
     * @param channel the channel
     * @throws IOException the io exception  这里异常需要处理
     */
    @RabbitListener(queues = {"FANOUT_QUEUE_A"})
    public void on(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        log.debug("FANOUT_QUEUE_A "+new String(message.getBody()));
    }

    /**
     * FANOUT广播队列监听二.
     *
     * @param message the message
     * @param channel the channel
     * @throws IOException the io exception   这里异常需要处理
     */
    @RabbitListener(queues = {"FANOUT_QUEUE_B"})
    public void t(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        log.debug("FANOUT_QUEUE_B "+new String(message.getBody()));
    }

    /**
     * DIRECT模式.
     *
     * @param message the message
     * @param channel the channel
     * @throws IOException the io exception  这里异常需要处理
     */
    @RabbitListener(queues = {"DIRECT_QUEUE"})
    public void message(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        log.debug("DIRECT "+new String (message.getBody()));
    }
}

编写发送消息接口来进行测试

package cn.felord.message.controller;

import cn.felord.message.bean.ResponseEntity;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * 消息接口.
 *
 * @author dax.
 * @version v1.0
 * @since 2018 /2/23 17:27
 */
@RestController
@RequestMapping("/rabbit")
public class SendController {
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     *  测试广播模式.
     *
     * @param p the p
     * @return the response entity
     */
    @RequestMapping("/fanout")
    public ResponseEntity send(String p) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        rabbitTemplate.convertAndSend("FANOUT_EXCHANGE", "", p, correlationData);
        return ResponseEntity.ok();
    }

    /**
     *  测试Direct模式.
     *
     * @param p the p
     * @return the response entity
     */
    @RequestMapping("/direct")
    public ResponseEntity direct(String p) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        rabbitTemplate.convertAndSend("DIRECT_EXCHANGE", "DIRECT_ROUTING_KEY", p, correlationData);
        return ResponseEntity.ok();
    }
}

测试广播模式

image

控制台输出

image

同样自己可以测试Direct模式可以打开rabbitmq控制台进行追踪相关运行信息

springboot 集成rabbitmq 并采用ack模式以

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

老男孩的成长之路

Java成长之路

springboot 集成rabbitmq 并采用ack模式 以

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

老男孩的成长之路

Java成长之路

springboot 集成rabbitmq 并采用ack模式以