ActiveMQ 客户端的开发

IT技术小咖

发布于 2019-06-26 15:49:44

7820

发布于 2019-06-26 15:49:44

文章被收录于专栏：码上修行码上修行

上篇文章 ActiveMQ 服务器的部署实现了 ActiveMQ 服务器的部署，本文分别以官方 API、Spring、SpringBoot 三种方式，实现 ActiveMQ 消息的生成者和消费者。

1.基于官方 API

ActiveMQ 官方实现了 JMS 接口，但使用很繁琐，不建议直接使用。 https://mvnrepository.com/artifact/org.apache.activemq/activemq-client

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
    <version>5.15.3</version>
</dependency>

2.基于 Spring 开发

采用 spring.xml 配置的方式，比 JMS 简单，但也比较繁琐。参见 《Spring实战》/ 12.Spring 消息

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
    <version>${spring-version}</version>
</dependency>

3.基于 SpringBoot 开发

IDEA 新建模块的 Spring Initializr 向导中，选中 I/O, JMS(ActiveMQ)；相应的 starter 是 spring-boot-starter-activemq

3.1 依赖包 pom.xml

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

3.2 配置文件 application.properties

# ActiveMQ 地址
spring.activemq.broker-url=tcp://centos:61616
# 参数 spring.jms.pub-sub-domain 用于指定消息模型是否为发布/订阅方式
# 默认情况下(false)，是点对占方式(queue)，如果要使用发布/订阅方式(topic)，必须设置为 true
spring.jms.pub-sub-domain=false

3.3 应用类 MessageApplication.java

其中定义了 4 个 Bean：

queue/topic: MessageProducer 中使用，作为消息发送的目标(分别是 queue/topic)；
containerFactoryQueue/containerFactoryTopic: MessageConsumer 中使用，以便同时启用两种消息模型。

需要特别说明的是：

只启用“点对点”模型：可配置 spring.jms.pub-sub-domain=false
只启用“发布/订阅”模型：可配置 spring.jms.pub-sub-domain=false
若要同时启用两种消息模型：必须定义 containerFactory Bean 并在消费者的 @JmsListener 中引用。

@SpringBootApplication
public class MessageApplication {
    @Bean
    public Queue queue() { // javax.jms.Queue
        return new ActiveMQQueue("queueName");
    }
    @Bean
    public Topic topic() { // javax.jms.Topic
        return new ActiveMQTopic("topicName");
    }
    @Bean
    public JmsListenerContainerFactory<?> containerFactoryQueue(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        bean.setConnectionFactory(connectionFactory);
        return bean;
    }
    @Bean
    public JmsListenerContainerFactory<?> containerFactoryTopic(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        bean.setConnectionFactory(connectionFactory);
        bean.setPubSubDomain(true);
        return bean;
    }
    public static void main(String[] args) {
        SpringApplication.run(MessageApplication.class, args);
    }
}

3.4 生产类 MessageProducer.java

@Component
@EnableScheduling
public class MessageProducer {
    @Resource
    private Queue queue; // javax.jms.Queue
    @Resource
    private Topic topic; // javax.jms.Topic
    @Resource
    private JmsMessagingTemplate messagingTemplate;
    @Scheduled(fixedDelay = 3000)
    public void sendQueue() { // 每隔 3 秒，发送一个“点对点”消息
        messagingTemplate.convertAndSend(queue, "Hi, queue!");
    }
    @Scheduled(fixedDelay = 6000)
    public void sendTopic() { // 每隔 6 秒，发送一个“发布/订阅”消息
        messagingTemplate.convertAndSend(topic, "Hi, topic!");
    }
}

3.5 消费类 MessageConsumer.java

@Component
public class MessageConsumer {
    private Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
    @JmsListener(destination = "queueName", containerFactory = "containerFactoryQueue")
    public void receiveQueue(String message) {
        logger.info("receiveQueue: " + message);
    }
    @JmsListener(destination = "topicName", containerFactory = "containerFactoryTopic")
    public void receiveTopic(String message) {
        logger.info("receiveTopic: " + message);
    }
}

3.6 运行结果

14:43:27.981  INFO  [MessageConsumer.java:22] - receiveTopic: Hi, topic!
14:43:27.993  INFO  [MessageConsumer.java:17] - receiveQueue: Hi, queue!
14:43:31.028  INFO  [MessageConsumer.java:17] - receiveQueue: Hi, queue!
14:43:34.008  INFO  [MessageConsumer.java:22] - receiveTopic: Hi, topic!
14:43:34.048  INFO  [MessageConsumer.java:17] - receiveQueue: Hi, queue!
14:43:37.169  INFO  [MessageConsumer.java:17] - receiveQueue: Hi, queue!

同时，可在 http://centos:8161/admin 中看到相关信息。

作者：王克锋

出处：https://kefeng.wang/2017/10/18/activemq-development/

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2019-05-13，如有侵权请联系 cloudcommunity@tencent.com 删除

java