项目中集成RabbitMQ配置，消息队列

小灰灰 2022-12-31 04:27 156阅读 0赞

1.依赖包

<dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.4.3</version>
            </dependency>

2.配置

@Configuration
    public class RabbitMqConfig {
        @Value(value = "${rabbit.host}")
        private String host;
        @Value(value = "${rabbit.port}")
        private int port;
        @Value(value = "${rabbit.username}")
        private String username;
        @Value(value = "${rabbit.password}")
        private String password;
        @Value(value = "${rabbit.virtualHost}")
        private String virtualHost;
        //交换机
        public static final String MIDDLE_USER_EXCHANGE = "user_exchange";
        //队列
        public static final String USER_NOTIFY = "user_register_queue";
        private static Connection connection;
    
        @Bean
        public ConnectionFactory connectionFactory() {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(host);
            factory.setPort(port);
            factory.setUsername(username);
            factory.setPassword(password);
            factory.setVirtualHost(virtualHost);
            return factory;
        }
    
        public static Connection getConnection(@Autowired ConnectionFactory factory) {
            try {
                if (connection != null && connection.isOpen()) {
                    return connection;
                }
                synchronized (RabbitMqConfig.class) {
                    if (connection != null && connection.isOpen()) {
                        return connection;
                    }
                    try {
                        connection = factory.newConnection();
                    } catch (IOException | TimeoutException up) {
                        throw new RuntimeException(up);
                    }
                    return connection;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }
    
        public static Channel getChannel(ConnectionFactory connectionFactory) {
            try {
                Channel channel = getConnection(connectionFactory).createChannel();
                //声明交换机
                channel.exchangeDeclare(MIDDLE_USER_EXCHANGE, "topic", true);
                channel.queueDeclare(USER_NOTIFY, true, false, false, null);
                //交换机绑定队列
                channel.queueBind(USER_NOTIFY, MIDDLE_USER_EXCHANGE, "");
                return channel;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

3.properties 配置

rabbit:
        host: 主机地址
        port: 5672
        username: 账号
        password: 密码
        virtualHost: /

4.具体用法

@Slf4j
    @Service
    public class NotifyServiceImpl implements INotifyService {
    
        @Autowired
        private ConnectionFactory connectionFactory;
    
        @Override
        public void userNotify(String userInfo) {
            Channel channel = null;
            try {
                log.info("用户队列通知参数：{}", userInfo);
                channel = RabbitMqConfig.getChannel(connectionFactory);
                channel.basicPublish(RabbitMqConfig.MIDDLE_USER_EXCHANGE, "", true, MessageProperties.PERSISTENT_TEXT_PLAIN, userInfo.getBytes());
            } catch (Exception e) {
                log.error("用户消息队列通知：{}", e);
            } finally {
                if (channel.isOpen()) {
                    try {
                        channel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    } catch (TimeoutException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

//队列通知
                    iNotifyService.userNotify(JSONObject.toJSONString(dto));