ActiveMQ
安装
下载网址:ActiveMQ
一定要和自己安装的jdk版本匹配,不然会报错
下载到本地之后解压缩
有可能端口号被占用
解除端口号占用,参考:Windows_端口被占用
打开cmd
查询所有的端口号
netstat -nao
查询指定端口号
netstat -ano|findstr 5672
查询什么程序在占用
tasklist | findstr "4756"
打开任务管理器看看这个程序
erl.exe是什么进程
Erlang 的执行程序,Erlang一种编程语言,多用于并发和分布式系统,现在最广泛使用在消息队列里面。
安装RabbitMQ的好像就是安装的这个,看来是RabbitMQ把这个端口给占用掉了,换个端口号
将5672改成55672
网址:http://localhost:8161/admin/
用户名:admin
密码:admin
成功。
使用
参考:消息队列之 ActiveMQ
Java访问ActiveMQ实例
引入依赖
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.2</version>
</dependency>
消息生产者
package mq.activeMQ;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class TopicPublisher {
/**
* 默认用户名
*/
public static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
/**
* 默认密码
*/
public static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
/**
* 默认连接地址
*/
public static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) {
//创建连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
try {
//创建连接
Connection connection = connectionFactory.createConnection();
//开启连接
connection.start();
//创建会话,不需要事务
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建 Topic,用作消费者订阅消息
Topic myTestTopic = session.createTopic("activemq-topic-test1");
//消息生产者
MessageProducer producer = session.createProducer(myTestTopic);
for (int i = 1; i <= 3; i++) {
TextMessage message = session.createTextMessage("发送消息 " + i);
producer.send(myTestTopic, message);
}
//关闭资源
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
消息消费者
package mq.activeMQ;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class TopicSubscriber {
/**
* 默认用户名
*/
public static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
/**
* 默认密码
*/
public static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
/**
* 默认连接地址
*/
public static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) {
//创建连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
try {
//创建连接
Connection connection = connectionFactory.createConnection();
//开启连接
connection.start();
//创建会话,不需要事务
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建 Topic
Topic myTestTopic = session.createTopic("activemq-topic-test1");
MessageConsumer messageConsumer = session.createConsumer(myTestTopic);
messageConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
System.out.println("消费者1 接收到消息:" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
MessageConsumer messageConsumer2 = session.createConsumer(myTestTopic);
messageConsumer2.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
System.out.println("消费者2 接收到消息:" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
MessageConsumer messageConsumer3 = session.createConsumer(myTestTopic);
messageConsumer3.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
System.out.println("消费者3 接收到消息:" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//让主线程休眠100秒,使消息消费者对象能继续存活一段时间从而能监听到消息
Thread.sleep(100 * 1000);
//关闭资源
session.close();
connection.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
gitee:JAVA集成AcitveMQ
启动ActiveMQ服务器
启动
Spring整合ActiveMQ
引入依赖
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.2</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>4.3.10.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.15.0</version>
</dependency>
Spring配置文件
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.0.xsd">
<!--这里要改base-package="基础包"-->
<!--下面还有要改的,看注释-->
<context:component-scan base-package="com.example.spring_activemq.activeMQ"/>
<!--连接池-->
<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL">
<value>tcp://localhost:61616</value>
</property>
</bean>
</property>
<property name="maxConnections" value="100"></property>
</bean>
<!--缓存-->
<bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="jmsFactory"/>
<property name="sessionCacheSize" value="1"/>
</bean>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!--获取连接、会话等对象-->
<property name="connectionFactory" ref="cachingConnectionFactory"/>
<!--消息转换器-->
<property name="messageConverter">
<bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
</property>
</bean>
<!--实际发送和接收消息的目的地-->
<bean id="testQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg name="name" value="spring-queue"/>
</bean>
<bean id="testTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-topic"/>
</bean>
<!--这里也要改class=""-->
<!--队列消息下的监视器-->
<bean id="queueListener" class="com.example.spring_activemq.activeMQ.QueueListener"/>
<!--主题模式下的接收器-->
<bean id="topic1Listener" class="com.example.spring_activemq.activeMQ.Topic1Listener"/>
<bean id="topic2Listener" class="com.example.spring_activemq.activeMQ.Topic2Listener"/>
<!--将消息监视器绑定到具体的消息目的地上-->
<bean id="queueContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="cachingConnectionFactory"/>
<property name="destination" ref="testQueue"/>
<property name="messageListener" ref="queueListener"/>
</bean>
<bean id="topic1Container"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<!--缓存-->
<property name="connectionFactory" ref="cachingConnectionFactory"/>
<!--目的地-->
<property name="destination" ref="testTopic"/>
<!--监视器-->
<property name="messageListener" ref="topic1Listener"/>
</bean>
<bean id="topic2Container"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="cachingConnectionFactory"/>
<property name="destination" ref="testTopic"/>
<property name="messageListener" ref="topic2Listener"/>
</bean>
</beans>
消息服务类
package com.example.spring_activemq.activeMQ;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.jms.*;
@Service
public class MessageService {
@Resource(name = "jmsTemplate")
private JmsTemplate jmsTemplate;
@Resource(name = "testQueue")
private Destination testQueue;
@Resource(name = "testTopic")
private Destination testTopic;
//向队列发送消息
public void sendQueueMessage(String messageContent) {
jmsTemplate.send(testQueue, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage msg = session.createTextMessage();
// 设置消息内容
msg.setText(messageContent);
return msg;
}
});
}
//向主题发送消息
public void sendTopicMessage(String messageContent) {
jmsTemplate.send(testTopic, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage msg = session.createTextMessage();
// 设置消息内容
msg.setText(messageContent);
return msg;
}
});
}
}
消息监听器类
队列监听器
package com.example.spring_activemq.activeMQ;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class QueueListener implements MessageListener {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
TextMessage txtMsg = (TextMessage) message;
String messageStr = txtMsg.getText();
System.out.println("队列监听器接收到文本消息:" + messageStr);
} catch (JMSException e) {
e.printStackTrace();
}
} else {
throw new IllegalArgumentException("只支持 TextMessage 类型消息!");
}
}
}
订阅消息监听器
package com.example.spring_activemq.activeMQ;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class Topic1Listener implements MessageListener {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
TextMessage txtMsg = (TextMessage) message;
String messageStr = txtMsg.getText();
System.out.println("主题监听器1 接收到文本消息:" + messageStr);
} catch (JMSException e) {
e.printStackTrace();
}
} else {
throw new IllegalArgumentException("只支持 TextMessage 类型消息!");
}
}
}
package com.example.spring_activemq.activeMQ;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class Topic2Listener implements MessageListener {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
TextMessage txtMsg = (TextMessage) message;
String messageStr = txtMsg.getText();
System.out.println("主题监听器2 接收到文本消息:" + messageStr);
} catch (JMSException e) {
e.printStackTrace();
}
} else {
throw new IllegalArgumentException("只支持 TextMessage 类型消息!");
}
}
}
启动应用
package com.example.spring_activemq.activeMQ;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class StartApplication {
public static void main(String[] args) {
//这里总是显示空指针,找半天也没找出来哪里错了,不找了,等以后学学再找。
ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:spring-context.xml");
MessageService messageService = (MessageService) ctx.getBean("messageService");
messageService.sendQueueMessage("我的测试消息1");
messageService.sendTopicMessage("我的测试消息2");
messageService.sendTopicMessage("我的测试消息3");
}
}
gitee:Spring集成ActiveMQ