RabbitMQ简单实例

2022-07-04

RabbitMQ简单实例

yaml配置文件

spring:
  # rabbitmq配置
  rabbitmq:
    host: xxx.xxx.xxx.xxx
    port: 5672
    username: xxx
    password: xxxx
    virtualHost: xxx-uat
    publisher-returns: true
    template.mandatory: true
    connection-timeout: 15000

配置文件RabbitMqConfig

import lombok.Data;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/** @description 消息队列配置 */
@Configuration
@ConfigurationProperties(prefix = "spring.rabbitmq")
@Data
public class RabbitMqConfig {

    private String host;
    private int port;
    private String userName;
    private String password;
    private String virtualHost;

    /**
     * 连接RabbitMQ，并加载到Bean
     *
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory cachingConnectionFactory =
                new CachingConnectionFactory(host, port);
        cachingConnectionFactory.setUsername(userName);
        cachingConnectionFactory.setPassword(password);
        cachingConnectionFactory.setVirtualHost(virtualHost);
        cachingConnectionFactory.setPublisherConfirms(true);
        return cachingConnectionFactory;
    }

    /**
     * RabbitMQ模板初始化，并加载到Bean
     *
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        return rabbitTemplate;
    }
}

常量RabbitMqConstants

/**
 * @description 消息队列定义的常量
 */
public class RabbitMqConstants {

    public static final String PAY_QUEUE = "queue.folkspace.pay";
    public static final String PAY_EXCHANGE = "exchange.folkspace.pay";
    public static final String PAY_ROUTING_KEY = "key.folkspace.pay";
    // 单位：ms
    public static final long PAY_DELAY = 5L;

}

队列 PayQueueConfig

package com.folkspace.pay.rabbitmq.queue;

import com.folkspace.pay.rabbitmq.constants.RabbitMqConstants;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @description 测试 消息队列
 */
@Configuration
public class PayQueueConfig {

    /**
     * 测试 延时触发事件
     *
     * @return
     */
    @Bean
    public CustomExchange payExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(
                RabbitMqConstants.PAY_EXCHANGE, "x-delayed-message", true, false, args);
    }

    /**
     * 测试 初始化消息队列
     *
     * @return
     */
    @Bean
    public Queue payQueue() {
        Queue queue = new Queue(RabbitMqConstants.PAY_QUEUE, true);
        return queue;
    }

    /**
     * 测试 绑定消息队列
     *
     * @return
     */
    @Bean
    public Binding payBinding() {
        return BindingBuilder.bind(payQueue())
                .to(payExchange())
                .with(RabbitMqConstants.PAY_ROUTING_KEY)
                .noargs();
    }
}

生产者PayMsgSender

import com.folkspace.pay.rabbitmq.constants.RabbitMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @description 测试sender 消息队列
 */
@Service
@Slf4j
public class PayMsgSender {

    @Autowired private RabbitTemplate rabbitTemplate;

    /**
     * 发送延迟事件
     *
     * @param msg 队列消息
     */
    public void sendPayMsg(String msg) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        log.info("支付消息发送时间: " + sdf.format(new Date()));
        rabbitTemplate.convertAndSend(
                RabbitMqConstants.PAY_EXCHANGE,
                RabbitMqConstants.PAY_ROUTING_KEY,
                msg,
                new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        message.getMessageProperties()
                                .setHeader("x-delay", RabbitMqConstants.PAY_DELAY);
                        return message;
                    }
                });
    }
}

消费者PayMsgReceiver

import com.folkspace.pay.rabbitmq.constants.RabbitMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.Date;

@Service
@Slf4j
public class PayMsgReceiver {

    /**
     * 监听超时触发事件
     *
     * @param msg
     */
    @RabbitListener(queues = RabbitMqConstants.PAY_QUEUE)
    public void receivePayMsg(String msg) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        log.info("支付消息接收时间: " + sdf.format(new Date()));
        log.info("支付接收到的消息: " + msg);
    }
}

测试类TestController

import com.folkspace.pay.rabbitmq.sender.PayIcbcCallbackAmountMsgSender;
import com.folkspace.pay.rabbitmq.sender.PayIcbcMsgSender;
import com.folkspace.pay.rabbitmq.sender.PayMsgSender;
import io.swagger.annotations.Api;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/test")
@Api(
        value = "TestController",
        tags = {"测试 API"})
public class TestController {

    @Autowired private PayMsgSender payMsgSender;

    @PostMapping("/sendMsg")
    public void sendMsg(String message) {
        payMsgSender.sendPayMsg(message);
    }

}