RabbitMQ常用配置

一、安装

使用docker 进行安装

docker pull rabbitmq:management
docker run -d --name rabbitmq89 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123456 -p 15672:15672 -p 5672:5672 rabbitmq:management

二、RabbitMQ 消息模型

在进行RabbitMQ消息模型之前需要在父项目中添加依赖。

 
    
        org.springframework.boot
        spring-boot-starter-parent
        2.3.8.RELEASE
    

    
    
        
            org.springframework.boot
            spring-boot-starter-web
        
        
        
            org.springframework.boot
            spring-boot-starter-amqp
        
        
            org.springframework.boot
            spring-boot-starter-test
            
                
                    org.junit.vintage
                    junit-vintage-engine

1、Work queues （1）yml配置

spring:
  rabbitmq:
    host: 192.xxx.xx.134
    port: 5672
    virtual-host: /
    username: admin
    password: xxxxxx

(2)编写生产者配置

@Component
public class WorkProducer {
 @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(){
        // 发送mq消息到消息队列中
        // 参数1：交换机。 如果是简单模型或工作队列模型，这里是空字符串；使用默认的交换机
        // 参数2：路由key。如果是简单模型或工作队列模型，这里传入队列名称
        // 参数3：消息内容
        rabbitTemplate.convertAndSend("","queue-simple","你好，世界");
    }


    // 创建队列
    @Bean
    public Queue createQueue(){
        return QueueBuilder.durable("queue-simple").build();
    }
    // 如果没有自动创建队列，需要配置RabbitAdmin
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }
}

（3）编写消费者

@Component
public class WorkConsumer {
    
    @RabbitListener(queuesToDeclare = @Queue(name = "queue-simple"))
    public void handlerMsg(String message) {
        System.out.println("消费者消费消息：" + message);
    }
}

2、fanout 交换机 (1)消费者

@Component
public class SeatConsumer {
    
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "queue-fanout-seat",durable = "true"),
            exchange = @Exchange(name = "exchange-fanout",type = ExchangeTypes.FANOUT)
    ))
    public void seat(String message) {
        System.out.println("消费者锁定座位：" + message);
    }
}

@Component
public class MemberConsumer {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "queue-fanout-member"),
            exchange = @Exchange(name = "exchange-fanout",type = ExchangeTypes.FANOUT)
    ))
    public void handlerMessage(String msg) {
        System.out.println("消费者添加积分："+msg);
    }
}

（3）生产者

@Component
public class FanoutProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String userId){
        rabbitTemplate.convertAndSend("exchange-fanout","","用户"+userId+"买了票");
    }
}

3、direct 交换机（1）消费者

@Component
public class ErrorConsumer {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "queue-direct-error"),
            exchange = @Exchange(name = "direct-log",type = ExchangeTypes.DIRECT),
            key = "error"
    ))
    public void handlerMessage(String msg) {
        System.out.println("消费者处理错误日志消息："+msg);
    }
}

（2）生产者

@Component
public class DirectProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendLog(String exchange,String routingKey,String msg){
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }
}

4、topic 交换机（1）消费者

@Component
public class C1 {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "Q1"),
            exchange = @Exchange(name = "exchange-topic",type = ExchangeTypes.TOPIC),
            key = "*.orange.*"
    ))
    public void handlerMsg(String msg) {
        System.out.println("消费者Q1：" + msg);
    }
}

@Component
public class C2 {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "Q2"),
            exchange = @Exchange(name = "exchange-topic",type = ExchangeTypes.TOPIC),
            key = {"*.*.rabbit","lazy.#"}
    ))
    public void handlerMsg(String msg) {
        System.out.println("消费者Q2：" + msg);
    }
}

（2）生产者

@Component
public class TopicProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void handlerMessage(String exchange,String routingKey,String msg) {
        rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }
}

5、RPC （1）服务端

@Component
public class Server {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "queue-rpc"),
            exchange = @Exchange(name = "exchange-rpc",type = ExchangeTypes.TOPIC),
            key = "rpc"
    ))
    public String handlerMsg(String msg) {
        System.out.println("服务端获取的数据：" + msg);
        return "服务端返回的数据";
    }
}

（2）客户端

@Component
public class ClientProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMsg(String exchange,String routingKey,String msg) {
        Object serverReturn = rabbitTemplate.convertSendAndReceive(exchange, routingKey, msg);
        System.out.println("serverReturn = " + serverReturn);
    }
}

6、PC （1）服务端消息可靠性保证

1.yml配置

spring:
  rabbitmq:
    host: 192.xxx.xx.134
    port: 5672
    virtual-host: /
    username: admin
    password: xxxxxx
    publisher-confirm-type: correlated

2.配置类

PostConstruct 注释用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化。此方法必须在将类放入服务之前调用。因为我在这个类中注入了其它服务，因此我使用这个注解在其它服务加载完成后进行初始化.

@Component
public class ConfirmProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RabbitConfirm rabbitConfirm;
    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(rabbitConfirm);
        // true 表示消息未到达队列时会把消息返回给生产者；false会直接丢弃消息
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(rabbitConfirm);
    }
    public void producerConfirm() {
        rabbitTemplate.convertAndSend("exchange-topic-confirm","confirm.msg","确认消息");
    }
}

@Component
public class RabbitConfirm implements
        RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("correlationData = " + correlationData);
        System.out.println("ack = " + ack); //false
        System.out.println("cause = " + cause);
        if (ack) {
            System.out.println("消息发送到交换机确认成功！");
        } else {
            System.out.println("消息发送到交换机失败，接下来根据业务做处理，比如：重发");
        }
    }
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        // 消息无法路由到队列时候，再这里可以获取消息内容
        System.out.println("=======消息回退：ReturnCallback=======");
        System.out.println("消息内容 = " + new String(message.getBody()));
        System.out.println("响应编码 = " + replyCode);
        System.out.println("消息无法路由的原因 = " + replyText);
        System.out.println("交换机 = " + exchange);
        System.out.println("路由key = " + routingKey);
    }
}

（2）消费端消息可靠性保证

1.yml配置

spring:
  rabbitmq:
    host: 192.168.12.134
    port: 5672
    username: admin
    password: 123456
    virtual-host: /
    listener:
      simple:
        retry:
          # 开启重试次数的限制，解决消费者出现异常死循环问题
          enabled: true
          # 最大重试次数
          max-attempts: 3
        # 开启消费者手动确认模式 （channel.bacisAck）
        acknowledge-mode: manual

2.配置类

死信配置：

@Component
public class DLQConsumer {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "queue-dlx"),
            exchange = @Exchange(name = "exchange-dlx",type = ExchangeTypes.TOPIC),
            key = "dlx.#"
    ))
    public void handlerMsg(String msg,Channel channel, Message message) throws IOException {
        System.out.println("死信队列处理消息：" + msg);
        // 消费后消息还在，因为没有确认。
    }
}

消费者配置：

@Component
public class ConfirmConsumer {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "queue-confirm",arguments = {
                    @Argument(name = "x-dead-letter-exchange",value = "exchange-dlx"),
                    @Argument(name = "x-dead-letter-routing-key",value = "dlx.ooo"),
                    @Argument(name = "x-message-ttl",value = "100000",type = "java.lang.Long"),
                    @Argument(name = "x-max-length",value = "3",type = "java.lang.Long"),
            }),
            exchange = @Exchange(name = "exchange-confirm",type = ExchangeTypes.TOPIC),
            key = "confirm.*"
    ))
    public void handlerMsg(String msg, Channel channel, Message message) throws IOException {
        try {
            //int i=1/0;
            System.out.println("消息:" + msg);
            // 消费者手动确认消息消费; 参数1：消息id；参数2：是否批量确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
        } catch (Exception e) {
            e.printStackTrace();
            // 需求：消息消费失败了，希望把消息重投到队列中，再次进行消费，如果还是消费失败就拒绝消息
            // 获取消息是否是重投的标记
            Boolean redelivered = message.getMessageProperties().getRedelivered();
            if (redelivered) {
                // 重投消息：可以拒绝不处理..
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("消息重投后消费消息失败，拒绝处理");
            } else {
                // 未重投过：重投
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,true);
                System.out.println("消息消费异常，进行重投！");
            }
        }
    }
}

三、其它配置 1、生产者创建交换机、队列（1）配置类

@Configuration
public class RabbitConfig {
    
    @Bean("confirmQueue")
    public Queue createConfirmQueue() {
        Map params = new HashMap<>();
        params.put("x-message-ttl",50000);
        params.put("x-dead-letter-exchange","exchange-dlx");
        params.put("x-dead-letter-routing-key","dlx.ooo");
        return new Queue("queue-confirm",true,false,false,params);
    }
    
    @Bean("exchangeConfirm")
    public Exchange exchangeConfirm(){
        return new TopicExchange("exchange-confirm",true,false);
    }
    
    @Bean
    public Binding bindConfirmQueue(
            @Qualifier("exchangeConfirm") Exchange exchange,
            @Qualifier("confirmQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange).with("confirm.#").noargs();
    }
    
    @Bean("dlxExchange")
    public Exchange dlxExchange(){
        return new TopicExchange("exchange-dlx",true,false);
    }
    
    @Bean("dlxQueue")
    public Queue createDLXQueue() {
        return new Queue("queue-dlx",true,false,false);
    }
    
    @Bean
    public Binding bindDlxQueue(
            @Qualifier("dlxExchange") Exchange exchange,
            @Qualifier("dlxQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange).with("dlx.#").noargs();
    }
}

2、发送接收对象（1）生产者

1.配置类

@Configuration
public class RabbitMQConfig {

    // 发送对象时自动转为json的配置类
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }


}

（2）消费者

下面这个注解写在要接收的对象上面

@Payload

添加配置类

@Configuration
public class RabbitMQConfig {

    // 发送对象时自动转为json的配置类
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }


}

RabbitMQ常用配置

Java相关栏目本月热门文章