ПО промежуточного слоя сообщений для начала работы с Spring Boot (5)

Введение

В промежуточном программном обеспечении сообщений есть две важные концепции: брокер сообщений и пункт назначения. Когда отправитель сообщения отправляет сообщение, сообщение принимается брокером сообщений, который гарантирует, что сообщение будет доставлено в указанное место назначения.

Наши часто используемые брокеры сообщений имеют спецификации JMS и AMQP. Соответственно, их распространенными реализациями являются ActiveMQ и RabbitMQ соответственно.

Предыдущая статья«Кэширование и NoSQL для начала работы с Spring Boot (4)».

2. Интегрируйте ActiveMQ

2.1 Добавить зависимости

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<!-- 如果需要配置连接池，添加如下依赖 -->
<dependency>  
    <groupId>org.apache.activemq</groupId>  
    <artifactId>activemq-pool</artifactId>  
</dependency>

2.2 Добавить конфигурацию

# activemq 配置
spring.activemq.broker-url=tcp://192.168.2.12:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.activemq.pool.enabled=false
spring.activemq.pool.max-connections=50
# 使用发布/订阅模式时，下边配置需要设置成 true
spring.jms.pub-sub-domain=false

Здесь spring.activemq.pool.enabled=false, что означает закрытие пула соединений.

2.3 Кодирование

Класс конфигурации:

@Configuration
public class JmsConfirguration {
    public static final String QUEUE_NAME = "activemq_queue";
    
    public static final String TOPIC_NAME = "activemq_topic";
    
    @Bean
    public Queue queue() {
        return new ActiveMQQueue(QUEUE_NAME);
    }
    
    @Bean
    public Topic topic() {
        return new ActiveMQTopic(TOPIC_NAME);
    }
}

Отвечает за создание очередей и тем.

Производитель сообщения:

@Component
public class JmsSender {
    @Autowired
    private Queue queue;
    
    @Autowired
    private Topic topic;
    
    @Autowired
    private JmsMessagingTemplate jmsTemplate;
    
    public void sendByQueue(String message) {
        this.jmsTemplate.convertAndSend(queue, message);
    }
    
    public void sendByTopic(String message) {
        this.jmsTemplate.convertAndSend(topic, message);
    }
}

Потребитель сообщения:

@Component
public class JmsReceiver {
    
    @JmsListener(destination = JmsConfirguration.QUEUE_NAME)
    public void receiveByQueue(String message) {
        System.out.println("接收队列消息:" + message);
    }
    
    @JmsListener(destination = JmsConfirguration.TOPIC_NAME)
    public void receiveByTopic(String message) {
        System.out.println("接收主题消息:" + message);
    }
}

Потребители сообщений используют аннотацию @JmsListener для прослушивания сообщений.

2.4 Тестирование

@RunWith(SpringRunner.class)
@SpringBootTest
public class JmsTest {
    @Autowired
    private JmsSender sender;
    @Test
    public void testSendByQueue() {
        for (int i = 1; i < 6; i++) {
            this.sender.sendByQueue("hello activemq queue " + i);
        }
    }
    
    @Test
    public void testSendByTopic() {
        for (int i = 1; i < 6; i++) {
            this.sender.sendByTopic("hello activemq topic " + i);
        }
    }
}

распечатать результат:

接收队列消息:hello activemq queue 1
接收队列消息:hello activemq queue 2
接收队列消息:hello activemq queue 3
接收队列消息:hello activemq queue 4
接收队列消息:hello activemq queue 5

При тестировании режима pub/sub установите spring.jms.pub-sub-domain=true

接收主题消息:hello activemq topic 1
接收主题消息:hello activemq topic 2
接收主题消息:hello activemq topic 3
接收主题消息:hello activemq topic 4
接收主题消息:hello activemq topic 5

3. Интегрируйте RabbitMQ

3.1 Добавить зависимости

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

3.2 Добавить конфигурацию

spring.rabbitmq.host=192.168.2.30
spring.rabbitmq.port=5672
spring.rabbitmq.username=light
spring.rabbitmq.password=light
spring.rabbitmq.virtual-host=/test

3.3 Кодирование

Класс конфигурации:

@Configuration
public class AmqpConfirguration {
    //=============简单、工作队列模式===============
    
    public static final String SIMPLE_QUEUE = "simple_queue";
    @Bean
    public Queue queue() {
        return new Queue(SIMPLE_QUEUE, true);
    }
    
    //===============发布/订阅模式============
    
    public static final String PS_QUEUE_1 = "ps_queue_1";
    public static final String PS_QUEUE_2 = "ps_queue_2";
    public static final String FANOUT_EXCHANGE = "fanout_exchange";
    
    @Bean
    public Queue psQueue1() {
        return new Queue(PS_QUEUE_1, true);
    }
    
    @Bean
    public Queue psQueue2() {
        return new Queue(PS_QUEUE_2, true);
    }
    
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE);
    }
    
    @Bean
    public Binding fanoutBinding1() {
        return BindingBuilder.bind(psQueue1()).to(fanoutExchange());
    }
    
    @Bean
    public Binding fanoutBinding2() {
        return BindingBuilder.bind(psQueue2()).to(fanoutExchange());
    }
    //===============路由模式============
    
    public static final String ROUTING_QUEUE_1 = "routing_queue_1";
    public static final String ROUTING_QUEUE_2 = "routing_queue_2";
    public static final String DIRECT_EXCHANGE = "direct_exchange";
    
    @Bean
    public Queue routingQueue1() {
        return new Queue(ROUTING_QUEUE_1, true);
    }
    
    @Bean
    public Queue routingQueue2() {
        return new Queue(ROUTING_QUEUE_2, true);
    }
    
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(DIRECT_EXCHANGE);
    }
    
    @Bean
    public Binding directBinding1() {
        return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("user");
    }
    
    @Bean
    public Binding directBinding2() {
        return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("order");
    }
    
    //===============主题模式============
    
    public static final String TOPIC_QUEUE_1 = "topic_queue_1";
    public static final String TOPIC_QUEUE_2 = "topic_queue_2";
    public static final String TOPIC_EXCHANGE = "topic_exchange";
    
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE_1, true);
    }
    
    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE_2, true);
    }
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    
    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.add");
    }
    
    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
    }
    
}

RabbitMQ имеет множество режимов работы, поэтому существует множество конфигураций. Читатели, которые хотят узнать больше о соответствующем контенте, могут проверить«Введение в рабочий режим RabbitMQ»Или информацию, связанную с Baidu, самостоятельно.

Производитель сообщения:

@Component
public class AmqpSender {
    @Autowired
    private AmqpTemplate amqpTemplate;
    /**
     * 简单模式发送
     * 
     * @param message
     */
    public void simpleSend(String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
    }
    /**
     * 发布/订阅模式发送
     * 
     * @param message
     */
    public void psSend(String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
    }
    /**
     * 路由模式发送
     * 
     * @param message
     */
    public void routingSend(String routingKey, String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
    }
    /**
     * 主题模式发送
     * 
     * @param routingKey
     * @param message
     */
    public void topicSend(String routingKey, String message) {
        this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
    }
}

Потребитель сообщения:

@Component
public class AmqpReceiver {
    /**
     * 简单模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
    public void simpleReceive(String message) {
        System.out.println("接收消息:" + message);
    }
    /**
     * 发布/订阅模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_1)
    public void psReceive1(String message) {
        System.out.println(AmqpConfirguration.PS_QUEUE_1 + "接收消息:" + message);
    }
    @RabbitListener(queues = AmqpConfirguration.PS_QUEUE_2)
    public void psReceive2(String message) {
        System.out.println(AmqpConfirguration.PS_QUEUE_2 + "接收消息:" + message);
    }
    /**
     * 路由模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_1)
    public void routingReceive1(String message) {
        System.out.println(AmqpConfirguration.ROUTING_QUEUE_1 + "接收消息:" + message);
    }
    @RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_2)
    public void routingReceive2(String message) {
        System.out.println(AmqpConfirguration.ROUTING_QUEUE_2 + "接收消息:" + message);
    }
    /**
     * 主题模式接收
     * 
     * @param message
     */
    @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_1)
    public void topicReceive1(String message) {
        System.out.println(AmqpConfirguration.TOPIC_QUEUE_1 + "接收消息:" + message);
    }
    
    @RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_2)
    public void topicReceive2(String message) {
        System.out.println(AmqpConfirguration.TOPIC_QUEUE_2 + "接收消息:" + message);
    }
}

Потребители сообщений используют аннотацию @RabbitListener для прослушивания сообщений.

3.4 Тестирование

@RunWith(SpringRunner.class)
@SpringBootTest
public class AmqpTest {
    @Autowired
    private AmqpSender sender;
    @Test
    public void testSimpleSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.simpleSend("test simpleSend " + i);
        }
    }
    @Test
    public void testPsSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.psSend("test psSend " + i);
        }
    }
    
    @Test
    public void testRoutingSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.routingSend("order", "test routingSend " + i);
        }
    }
    
    @Test
    public void testTopicSend() {
        for (int i = 1; i < 6; i++) {
            this.sender.topicSend("user.add", "test topicSend " + i);
        }
    }
}

Результаты теста пропускаются. . .

Напоминание о наступлении на яму 1: ACCESS_REFUSED — вход в систему был отклонен с использованием механизма аутентификации PLAIN

решение:

1) Пожалуйста, убедитесь, что имя пользователя и пароль верны.Следует отметить, содержит ли значение имени пользователя и пароля пробелы или табуляции (когда автор тестировал это, аутентификация не удалась, потому что в пароле была лишняя табуляция).

2) Если тестовая учетная запись использует гостя, вам необходимо изменить файл rabbitmq.conf. Добавьте в этот файл конфигурацию «loopback_users=none».

Напоминание о яме 2: Невозможно подготовить очередь для слушателя. Либо очередь не существует, либо брокер не разрешает нам ее использовать.

решение:

Мы можем войти в интерфейс управления RabbitMQ и вручную добавить соответствующую очередь в параметре Queue.

4. Ссылки

• Краткое введение в промежуточное программное обеспечение сообщений

• Официальная документация Spring Boot

• Связанный с контролем доступа Rabbit MQ

• Автор этой статьи: moonlightL
• Ссылка на эту статью: Woohoo.ext light.com/2018/01/26/…
• Уведомление об авторских правах:Все статьи в этом блоге являются оригинальными, если не указано иное.CC BY-NC-SA 4.0соглашение. Для перепечатки, пожалуйста, укажите исходную ссылку и автора и другую соответствующую информацию в четкой позиции в начале статьи, четко укажите изменение (если таковое имеется) и сообщите об этом по электронной почте и другими способами, спасибо за сотрудничество !