ActiveMQ

xiaoxiao2021-03-01 7

1.3.1下载

官方网站下载：http://activemq.apache.org/

1.3.2安装（Linux）

将apache-activemq-5.12.0-bin.tar.gz 上传至服务器解压此文件

tar zxvf apache-activemq-5.12.0-bin.tar.gz

为apache-activemq-5.12.0目录赋权

chmod 777 apache-activemq-5.12.0

进入apache-activemq-5.12.0\bin目录

（5）赋与执行权限

chmod 755 activemq

-------------------------------------- 知识点小贴士 --------------------------

linux 命令chmod 755的意思

chmod是Linux下设置文件权限的命令，后面的数字表示不同用户或用户组的权限。

一般是三个数字：第一个数字表示文件所有者的权限第二个数字表示与文件所有者同属一个用户组的其他用户的权限第三个数字表示其它用户组的权限。

权限分为三种：读（r=4），写（w=2），执行（x=1）。综合起来还有可读可执行（rx=5=4+1）、可读可写（rw=6=4+2）、可读可写可执行(rwx=7=4+2+1)。

所以，chmod 755 设置用户的权限为：

1.文件所有者可读可写可执行 --7

2.与文件所有者同属一个用户组的其他用户可读可执行 --5 3.其它用户组可读可执行 --5

1.3.3启动

./activemq start

出现下列提示表示成功！

假设服务器地址为192.168.25.135 ，打开浏览器输入地址

http://192.168.25.135:8161/ 即可进入ActiveMQ管理页面

点击进入管理页面

输入用户名和密码均为 admin

进入主界面

点对点消息列表：

列表各列信息含义如下：

Number Of Pending Messages ：等待消费的消息这个是当前未出队列的数量。

Number Of Consumers ：消费者这个是消费者端的消费者数量

Messages Enqueued ：进入队列的消息进入队列的总数量,包括出队列的。

Messages Dequeued ：出了队列的消息可以理解为是消费这消费掉的数量。

2. JMS入门小Demo

2.1点对点模式

点对点的模式主要建立在一个队列上面，当连接一个列队的时候，发送端不需要知道接收端是否正在接收，可以直接向ActiveMQ发送消息，发送的消息，将会先进入队列中，如果有接收端在监听，则会发向接收端，如果没有接收端接收，则会保存在activemq服务器，直到接收端接收消息，点对点的消息模式可以有多个发送端，多个接收端，但是一条消息，只会被一个接收端给接收到，哪个接收端先连上ActiveMQ，则会先接收到，而后来的接收端则接收不到那条消息。

2.1.1消息生产者

（1）创建工程jmsDemo ，引入依赖

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-client</artifactId>

</dependency>

（2）创建类QueueProducer main方法代码如下：

//1.创建连接工厂

ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");

//2.获取连接

Connection connection = connectionFactory.createConnection();

//3.启动连接

connection.start();

//4.获取session (参数1：是否启动事务,参数2：消息确认模式)

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//5.创建队列对象

Queue queue = session.createQueue("test-queue");

//6.创建消息生产者

MessageProducer producer = session.createProducer(queue);

//7.创建消息

TextMessage textMessage = session.createTextMessage("欢迎来到神奇的品优购世界");

//8.发送消息

producer.send(textMessage);

//9.关闭资源

producer.close();

session.close();

connection.close();

上述代码中第4步创建session 的两个参数：

第1个参数是否使用事务

第2个参数消息的确认模式

AUTO_ACKNOWLEDGE = 1 自动确认CLIENT_ACKNOWLEDGE = 2 客户端手动确认 DUPS_OK_ACKNOWLEDGE = 3 自动批量确认SESSION_TRANSACTED = 0 事务提交并确认

运行后通过ActiveMQ管理界面查询

2.1.2消息消费者

创建类QueueConsumer ，main方法代码如下：

//1.创建连接工厂

ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");

//2.获取连接

Connection connection = connectionFactory.createConnection();

//3.启动连接

connection.start();

//4.获取session (参数1：是否启动事务,参数2：消息确认模式)

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//5.创建队列对象

Queue queue = session.createQueue("test-queue");

//6.创建消息消费

MessageConsumer consumer = session.createConsumer(queue);

//7.监听消息

consumer.setMessageListener(new MessageListener() {

public void onMessage(Message message) {

TextMessage textMessage=(TextMessage)message;

try {

System.out.println("接收到消息："+textMessage.getText());

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

});

//8.等待键盘输入

System.in.read();

//9.关闭资源

consumer.close();

session.close();

connection.close();

执行后看到控制台输出

2.1.3运行测试

同时开启2个以上的消费者，再次运行生产者，观察每个消费者控制台的输出，会发现只有一个消费者会接收到消息。

2.2 发布/订阅模式

2.2.1消息生产者

创建类TopicProducer ，main方法代码如下：

//1.创建连接工厂

ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");

//2.获取连接

Connection connection = connectionFactory.createConnection();

//3.启动连接

connection.start();

//4.获取session (参数1：是否启动事务,参数2：消息确认模式)

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//5.创建主题对象

Topic topic = session.createTopic("test-topic");

//6.创建消息生产者

MessageProducer producer = session.createProducer(topic);

//7.创建消息

TextMessage textMessage = session.createTextMessage("欢迎来到神奇的品优购世界");

//8.发送消息

producer.send(textMessage);

//9.关闭资源

producer.close();

session.close();

connection.close();

运行效果如下：

2.2.2消息消费者

创建类TopicConsumer ,main方法代码如下：

//1.创建连接工厂

ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");

//2.获取连接

Connection connection = connectionFactory.createConnection();

//3.启动连接

connection.start();

//4.获取session (参数1：是否启动事务,参数2：消息确认模式)

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//5.创建主题对象

//Queue queue = session.createQueue("test-queue");

Topic topic = session.createTopic("test-topic");

//6.创建消息消费

MessageConsumer consumer = session.createConsumer(topic);

//7.监听消息

consumer.setMessageListener(new MessageListener() {

public void onMessage(Message message) {

TextMessage textMessage=(TextMessage)message;

try {

System.out.println("接收到消息："+textMessage.getText());

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

});

//8.等待键盘输入

System.in.read();

//9.关闭资源

consumer.close();

session.close();

connection.close();

2.2.3运行测试

同时开启2个以上的消费者，再次运行生产者，观察每个消费者控制台的输出，会发现每个消费者会接收到消息。

3.Spring整合JMS

3.1 点对点模式

3.1.1消息生产者

创建工程springjms_producer，在POM文件中引入SpringJms 、activeMQ以及单元测试相关依赖

（2）在src/main/resources下创建spring配置文件applicationContext-jms-producer.xml

<context:component-scan base-package="cn.itcast.demo"></context:component-scan>

</bean>

</bean>

</bean>

<constructor-arg value="queue_text"/>

</bean>

（3）在cn.itcast.demo包下创建消息生产者类

@Component

public class QueueProducer {

@Autowired

private JmsTemplate jmsTemplate;

@Autowired

private Destination queueTextDestination;

/**

* 发送文本消息

* @param text

public void sendTextMessage(final String text){

jmsTemplate.send(queueTextDestination, new MessageCreator() {

public Message createMessage(Session session) throws JMSException {

return session.createTextMessage(text);

}

});

}

（4）单元测试

在src/test/java创建测试类

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration(locations="classpath:applicationContext-jms-producer.xml")

public class TestQueue {

@Autowired

private QueueProducer queueProducer;

@Test

public void testSend(){

queueProducer.sendTextMessage("SpringJms-点对点");

}

3.1.2消息消费者

创建工程springjms_consumer，在POM文件中引入依赖（同上一个工程）

（2）创建配置文件 applicationContext-jms-consumer-queue.xml

</bean>

</bean>

<constructor-arg value="queue_text"/>

</bean>

</bean>

（3）编写监听类

public class MyMessageListener implements MessageListener {

public void onMessage(Message message) {

TextMessage textMessage=(TextMessage)message;

try {

System.out.println("接收到消息："+textMessage.getText());

} catch (JMSException e) {

e.printStackTrace();

}

（4）创建测试类

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration(locations="classpath:applicationContext-jms-consumer-queue.xml")

public class TestQueue {

@Test

public void testQueue(){

try {

System.in.read();

} catch (IOException e) {

e.printStackTrace();

}

3.2发布/订阅模式

3.2.1消息生产者

（1）在工程springjms_producer的applicationContext-jms-producer.xml增加配置

<constructor-arg value="topic_text"/>

</bean>

（2）创建生产者类

@Component

public class TopicProducer {

@Autowired

private JmsTemplate jmsTemplate;

@Autowired

private Destination topicTextDestination;

/**

* 发送文本消息

* @param text

public void sendTextMessage(final String text){

jmsTemplate.send(topicTextDestination, new MessageCreator() {

public Message createMessage(Session session) throws JMSException {

return session.createTextMessage(text);

}

});

}

（3）编写测试类

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.test.context.ContextConfiguration;

import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import cn.itcast.demo.TopicProducer;

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration(locations="classpath:applicationContext-activemq-producer.xml")

public class TestTopic {

@Autowired

private TopicProducer topicProducer;

@Test

public void sendTextQueue(){

topicProducer.sendTextMessage();

}

3.2.2消息消费者

在activemq-spring-consumer工程中创建配置文件applicationContext-jms-consumer-topic.xml

</bean>

</bean>

<constructor-arg value="topic_text"/>

</bean>

</bean>

编写测试类

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration(locations="classpath:applicationContext-jms-consumer-topic.xml")

public class TestTopic {

@Test

public void testTopic(){

try {

System.in.read();

} catch (IOException e) {

e.printStackTrace();

}

测试：同时运行三个消费者工程，在运行生产者工程，查看三个消费者工程的控制台输出。

4.商品审核-导入Solr索引库

4.1需求分析

运用消息中间件activeMQ实现运营商后台与搜索服务的零耦合。运营商执行商品审核后，向activeMQ发送消息（SKU列表），搜索服务从activeMQ接收到消息并导入到solr索引库。

4.2消息生产者（运营商后台）

4.2.1解除耦合

修改pinyougou-manager-web，移除搜索服务接口依赖：

<groupId>com.pinyougou</groupId>

<artifactId>pinyougou-search-interface</artifactId>

<version>0.0.3-SNAPSHOT</version>

</dependency>

GoodsController.java中删除调用搜索服务接口的相关代码

//itemSearchService.deleteByGoodsIds(Arrays.asList(ids));

//@Reference

//private ItemSearchService itemSearchService;

//itemSearchService.importList(itemList);

4.2.2准备工作

（1）修改pinyougou-manager-web的pom.xml,引入依赖

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-client</artifactId>

</dependency>

pinyougou-sellergoods-service工程添加applicationContext-jms-producer.xml

改名为spring-activemq.xml 内容如下：

</bean>

</bean>

</bean>

<constructor-arg value="pinyougou_queue_solr"/>

</bean>

（3）修改web.xml

<context-param>

<param-name>contextConfigLocation</param-name>

<param-value>classpath:spring/spring-*.xml</param-value>

</context-param>

4.2.3代码实现

修改pinyougou-manager-web工程的GoodsController.java

@Autowired

private Destination queueSolrDestination;//用于发送solr导入的消息

@Autowired

private JmsTemplate jmsTemplate;

@RequestMapping("/updateStatus")

public Result updateStatus(Long[] ids,String status){

try {

goodsService.updateStatus(ids, status);

//按照SPU ID查询 SKU列表(状态为1)

if(status.equals("1")){//审核通过

List<TbItem> itemList = goodsService.findItemListByGoodsIdandStatus(ids, status);

//调用搜索接口实现数据批量导入

if(itemList.size()>0){

final String jsonString = JSON.toJSONString(itemList);

jmsTemplate.send(queueSolrDestination, new MessageCreator() {

@Override

public Message createMessage(Session session) throws JMSException {

return session.createTextMessage(jsonString);

}

});

}else{

System.out.println("没有明细数据");

}

return new Result(true, "修改状态成功");

} catch (Exception e) {

e.printStackTrace();

return new Result(false, "修改状态失败");

}

4.3消息消费者（搜索服务）

4.3.1准备工作

（1）修改pinyougou-search-service ，在pom.xml中添加activemq依赖

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-client</artifactId>

</dependency>

（2）添加spring配置文件applicationContext-jms-consumer.xml

</bean>

</bean>

<constructor-arg value="pinyougou_queue_solr"/>

</bean>

</bean>

4.3.2代码实现

在pinyougou-search-service的com.pinyougou.search.service.impl新增监听类

@Component

public class ItemSearchListener implements MessageListener{

@Autowired

private ItemSearchService itemSearchService;

@Override

public void onMessage(Message message) {

System.out.println("监听接收到消息...");

try {

TextMessage textMessage=(TextMessage)message;

String text = textMessage.getText();

List<TbItem> list = JSON.parseArray(text,TbItem.class);

for(TbItem item:list){

System.out.println(item.getId()+" "+item.getTitle());

Map specMap= JSON.parseObject(item.getSpec());//将spec字段中的json字符串转换为map

item.setSpecMap(specMap);//给带注解的字段赋值

}

itemSearchService.importList(list);//导入

System.out.println("成功导入到索引库");

} catch (Exception e) {

e.printStackTrace();

}

5.商品删除-移除Solr索引库记录

5.1需求分析

通过消息中间件实现在商品删除时也同时移除索引库记录的功能。

5.2消息生产者（运营商后台）

5.2.1配置文件

修改pinyougou-manager-web工程的spring-activemq.xml，添加bean配置

<constructor-arg value="pinyougou_queue_solr_delete"/>

</bean>

5.2.2代码实现

修改GoodsController.java

@Autowired

private Destination queueSolrDeleteDestination;//用户在索引库中删除记录

/**

* 批量删除

* @param ids

* @return

@RequestMapping("/delete")

public Result delete(final Long [] ids){

try {

goodsService.delete(ids);

jmsTemplate.send(queueSolrDeleteDestination, new MessageCreator() {

@Override

public Message createMessage(Session session) throws JMSException {

return session.createObjectMessage(ids);

}

});

return new Result(true, "删除成功");

} catch (Exception e) {

e.printStackTrace();

return new Result(false, "删除失败");

}

5.3消息消费者（搜索服务）

5.3.1配置文件

修改pinyougou-search-service的applicationContext-activemq-consumer.xml

<constructor-arg value="pinyougou_queue_solr_delete"/>

</bean>

</bean>

5.3.2代码实现

com.pinyougou.search.service.impl包下创建监听类ItemDeleteListener

/**

* 监听：用于删除索引库中记录

* @author Administrator

@Component

public class ItemDeleteListener implements MessageListener{

@Autowired

private ItemSearchService itemSearchService;

@Override

public void onMessage(Message message) {

try {

ObjectMessage objectMessage= (ObjectMessage)message;

Long[] goodsIds = (Long[]) objectMessage.getObject();

System.out.println("ItemDeleteListener监听接收到消息..."+goodsIds);

itemSearchService.deleteByGoodsIds(Arrays.asList(goodsIds));

System.out.println("成功删除索引库中的记录");

} catch (Exception e) {

e.printStackTrace();

}

6.商品审核-执行网页静态化

6.1需求分析

运用消息中间件activeMQ实现运营商后台与网页生成服务的零耦合。运营商执行商品审核后，向activeMQ发送消息（商品ID），网页生成服务从activeMQ接收到消息后执行网页生成操作。

6.2消息生产者（运营商后台）

6.2.1解除耦合

修改pinyougou-manager-web，移除网页生成服务接口依赖：

<groupId>com.pinyougou</groupId>

<artifactId>pinyougou-page-interface</artifactId>

<version>0.0.3-SNAPSHOT</version>

</dependency>

GoodsController.java中删除调用网页生成服务接口的相关代码

//private ItemPageService itemPageService;

//静态页生成

//for(Long goodsId:ids){

//itemPageService.genItemHtml(goodsId);

//}

6.2.2准备工作

修改配置文件spring-activemq.xml，添加配置

<constructor-arg value="pinyougou_topic_page"/>

</bean>

6.2.3代码实现

修改pinyougou-manager-web的GoodsController.java

public Result updateStatus(Long[] ids,String status){

try {

.......

if(status.equals("1")){//审核通过

........

//静态页生成

for(final Long goodsId:ids){

jmsTemplate.send(topicPageDestination, new MessageCreator() {

@Override

public Message createMessage(Session session) throws JMSException {

return session.createTextMessage(goodsId+"");

}

});

}

......

} catch (Exception e) {

......

}

6.3消息消费者（页面生成服务）

6.3.1解除dubbox依赖

修改工程pinyougou-page-service ，删除dubbox相关依赖

<groupId>com.alibaba</groupId>

<artifactId>dubbo</artifactId>

</dependency>

<groupId>org.apache.zookeeper</groupId>

<artifactId>zookeeper</artifactId>

</dependency>

<groupId>com.github.sgroschupf</groupId>

<artifactId>zkclient</artifactId>

</dependency>

修改applicationContext-service.xml，删除dubbox相关配置

<dubbo:protocol name="dubbo" port="20885"></dubbo:protocol>

<dubbo:application name="pinyougou-page-service"/>

<dubbo:registry address="zookeeper://192.168.25.135:2181"/>

<dubbo:annotation package="com.pinyougou.page.service.impl" />

修改ItemPageServiceImpl类的@Service注解为org.springframework.stereotype.Service包下的@Service注解

6.3.2准备工作

（1）修改applicationContext-service.xml，添加配置

<context:component-scan base-package="com.pinyougou.page.service.impl"/>

（2）pom.xml中引入activeMQ客户端的依赖

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-client</artifactId>

</dependency>

（3）添加spring配置文件applicationContext-jms-consumer.xml

</bean>

</bean>

<constructor-arg value="pinyougou_queue_page"/>

</bean>

</bean>

6.3.3代码编写

创建消息监听类PageListener

@Component

public class PageListener implements MessageListener {

@Autowired

private ItemPageService itemPageService;

@Override

public void onMessage(Message message) {

TextMessage textMessage= (TextMessage)message;

try {

String text = textMessage.getText();

System.out.println("接收到消息："+text);

boolean b = itemPageService.genItemHtml(Long.parseLong(text));

} catch (Exception e) {

e.printStackTrace();

}

7.商品删除-删除商品详细页

7.1需求分析

执行商品删除后，同时删除每个服务器上的商品详细页

7.2消息生产者（运营商后台）

7.2.1配置文件

修改spring-activemq.xml，添加配置

<constructor-arg value="pinyougou_topic_page_delete"/>

</bean>

7.2.2代码实现

修改GoodsController.java

@Autowired

private Destination topicPageDeleteDestination;//用于删除静态网页的消息

/**

* 批量删除

* @param ids

* @return

@RequestMapping("/delete")

public Result delete(final Long [] ids){

try {

goodsService.delete(ids);

jmsTemplate.send(queueSolrDeleteDestination, new MessageCreator() {

@Override

public Message createMessage(Session session) throws JMSException {

return session.createObjectMessage(ids);

}

});

//删除页面

jmsTemplate.send(topicPageDeleteDestination, new MessageCreator() {

@Override

public Message createMessage(Session session) throws JMSException {

return session.createObjectMessage(ids);

}

});

return new Result(true, "删除成功");

} catch (Exception e) {

e.printStackTrace();

return new Result(false, "删除失败");

}

7.3消息消费者（页面生成服务）

7.3.1配置文件

修改pinyougou-page-service的applicationContext-activemq-consumer.xml

<constructor-arg value="pinyougou_topic_page_delete"/>

</bean>

</bean>

7.3.2代码实现

（1）修改pinyougou-page-interface的ItemPageService.java

/**

* 删除商品详细页

* @param goodsId

* @return

public boolean deleteItemHtml(Long[] goodsIds);

修改pinyougou-page-service的ItemPageServiceImpl.java

@Override

public boolean deleteItemHtml(Long[] goodsIds) {

try {

for(Long goodsId:goodsIds){

new File(pagedir+goodsId+".html").delete();

}

return true;

} catch (Exception e) {

e.printStackTrace();

return false;

}

创建监听类PageDeleteListener

@Component

public class PageDeleteListener implements MessageListener {

@Autowired

private ItemPageService itemPageService;

@Override

public void onMessage(Message message) {

ObjectMessage objectMessage= (ObjectMessage)message;

try {

Long[] goodsIds = (Long[]) objectMessage.getObject();

System.out.println("ItemDeleteListener监听接收到消息..."+goodsIds);

boolean b = itemPageService.deleteItemHtml(goodsIds);

System.out.println("网页删除结果："+b);

} catch (JMSException e) {

e.printStackTrace();

}

转载请注明原文地址: https://www.6miu.com/read-3100155.html

技术

最新回复(0)