消息中间件-ActivityMQ系列文章-入门及例子

文章目录

ActivityMQ安装
入门例子
- pom.xml
- MQ配置
- 消息创建端
- 消息消费端

ActivityMQ安装

前提：安装JDK

上官网下载apache-activemq-5.16.0windows版本，解压。
进入apache-activemq-5.16.0\bin\win64 双击ctivemq.bat
窗口打印如下信息：

Size=104857600
jvm 1    |  INFO | Connector mqtt started
jvm 1    |  INFO | Starting Jetty server
jvm 1    |  INFO | Creating Jetty connector
jvm 1    |  WARN | ServletContext@o.e.j.s.ServletContextHandler@76a311d9{/,null,STARTING} has uncovered http methods for path: /
jvm 1    |  INFO | Listening for connections at ws://DESKTOP-HLLDRH0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1    |  INFO | Connector ws started
jvm 1    |  INFO | Apache ActiveMQ 5.16.0 (localhost, ID:DESKTOP-HLLDRH0-50964-1606466800157-0:1) started
jvm 1    |  INFO | For help or more information please see: http://activemq.apache.org
jvm 1    |  INFO | ActiveMQ WebConsole available at http://127.0.0.1:8161/
jvm 1    |  INFO | ActiveMQ Jolokia REST API available at http://127.0.0.1:8161/api/jolokia/

启动成功

浏览器上输入：http://127.0.0.1:8161/index.html
输入用户名密码admin/admin 默认

点击画圈的链接。

根据英文单词能看出菜单的含义。队列，主题、链接、网络等。做例子的时候可以点击看看具体里面是什么。

入门例子

需要编写一个消息生产端和一个消息消费方，达到消息的传递模拟。

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.4.0</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.ycxy</groupId><artifactId>producer</artifactId><version>0.0.1-SNAPSHOT</version><name>producer</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-pool --><!-- 使用springboot2.0+及以下版本时候，maven配置依赖是： --><!-- <dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.16.0</version></dependency>--><!--使用springboot2.1+时候，maven配置依赖是 --><dependency><groupId>org.messaginghub</groupId><artifactId>pooled-jms</artifactId><version>1.2.1</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

重点说明：看里面的注释，与 ConnectionFactory和JmsMessagingTemplate注入有关。使用了pooled-jms 之后这两个类就默认定义的Bean，可以直接使用，否则需要显示的自己创建。

MQ配置

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.config.SimpleJmsListenerContainerFactory;
import org.springframework.jms.core.JmsMessagingTemplate;import javax.jms.ConnectionFactory;
import javax.jms.Queue;
import javax.jms.Topic;@Configuration
public class MQconfig {@Value("${spring.activemq.broker-url}")private String brokerUrl;@Value("${spring.activemq.user}")private String username;@Value("${spring.activemq.topic-name}")private String password;@Value("${spring.activemq.queue-name}")private String queueName;@Value("${spring.activemq.topic-name}")private String topicName;@Bean(name = "queue")public Queue queue() {return new ActiveMQQueue(queueName);}@Bean(name = "topic")public Topic topic() {return new ActiveMQTopic(topicName);}//    @Bean
//    public ConnectionFactory connectionFactory(){//        return new ActiveMQConnectionFactory(username, password, brokerUrl);
//    }
//
//    @Bean
//    public JmsMessagingTemplate jmsMessageTemplate(){//        return new JmsMessagingTemplate(connectionFactory());
//    }// 在Queue模式中，对消息的监听需要对containerFactory进行配置@Bean("queueListener")public JmsListenerContainerFactory<?> queueJmsListenerContainerFactory(ConnectionFactory connectionFactory){SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setPubSubDomain(false);return factory;}//在Topic模式中，对消息的监听需要对containerFactory进行配置@Bean("topicListener")public JmsListenerContainerFactory<?> topicJmsListenerContainerFactory(ConnectionFactory connectionFactory){SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setPubSubDomain(true);return factory;}
}

这段在生产端和消费端都可以使用，配置以这样。Spring封装了对消息的操作，主要使用JmsMessagingTemplate进行操作，可以参考《Spring实战》一书。

消息创建端

我写了一个RestController用于测试消息产生。核心调用jmsMessagingTemplate.convertAndSend(destination, message);把消息产生之后发送给MQ哪个地方。

import javax.jms.Destination;
import javax.jms.Queue;
import javax.jms.Topic;@RestController
public class ProducerController {//连接池jmspooled自动创建bean@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;@Autowiredprivate Queue queue;@Autowiredprivate Topic topic;@PostMapping("/queue/test")public String sendQueue(@RequestBody String str) {this.sendMessage(this.queue, str);return "success";}@PostMapping("/topic/test")public String sendTopic(@RequestBody String str) {this.sendMessage(this.topic, str);return "success";}// 发送消息，destination是发送到的队列，message是待发送的消息private void sendMessage(Destination destination, final String message){jmsMessagingTemplate.convertAndSend(destination, message);}
}

消息消费端

创建一个监听用于监听指定的队列，接受到消息之后调用具体业务逻辑进行处理。

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;@Component
public class QueueConsumerListener
{//queue模式的消费者@JmsListener(destination="${spring.activemq.queue-name}", containerFactory="queueListener")public void readActiveQueue(String message) {System.out.println("queue接受到：" + message);}
}

@Component
public class TopicConsumerListener
{//topic模式的消费者@JmsListener(destination="${spring.activemq.topic-name}", containerFactory="topicListener")public void readActiveQueue(String message) {System.out.println("topic接受到：" + message);}
}