ActiveMQ 使用
Posted zhenghuasheng
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ActiveMQ 使用相关的知识，希望对你有一定的参考价值。
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。
主要特点：
多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, php。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性
通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上
支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
支持通过JDBC和journal提供高速的消息持久化
从设计上保证了高性能的集群,客户端-服务器,点对点
支持Ajax
支持与Axis的整合
可以很容易得调用内嵌JMS provider,进行测试
因为测试在本机搭建，使用的是Windows版本来演示，官网下载地址：http://activemq.apache.org/activemq-5141-release.html
安装步骤为Windows版本，linux版本安装大致相同，在官网有明确说明步骤
下载后为：apache-activemq-5.13.4-bin.zip的压缩包，解压后如图：
apache-activemq-5.13.4\\bin\\win64\\activemq.bat 为启动脚本
双击启动MQ，启动默认端口为8161
修改启动端口在apache-activemq-5.13.4\\conf\\jetty.xml
可见activemq使用jetty启动 ,启动完成后，后台管理界面可以访问：http://localhost:8161/
点击manager ActiveMQ broker时会提示输入用户名和密码
可以设置apache-activemq-5.13.4\\conf\\jetty.xml中<property name="authenticate" value="true" />为false，避免登录验证
<bean id="securityConstraint" class="org.eclipse.jetty.util.security.Constraint">
<property name="name" value="BASIC" />
<property name="roles" value="user,admin" />
<!-- set authenticate=false to disable login -->
<property name="authenticate" value="true" />
</bean>
<bean id="adminSecurityConstraint" class="org.eclipse.jetty.util.security.Constraint">
<property name="name" value="BASIC" />
<property name="roles" value="admin" />
<!-- set authenticate=false to disable login -->
<property name="authenticate" value="true" />
</bean>也可以修改登录密码：
apache-activemq-5.13.4\\conf\\jetty-realm.properties
# Defines users that can access the web (console, demo, etc.)
# username: password [,rolename ...]
admin: zhs, admin
user: user, useractiveMQ默认采用tcp协议连接，连接端口为：61616，支持多中协议，也可以更改配置
<transportConnectors>
<!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>测试代码：
producer
**
* 消息的生产者（发送者）
*
* @author zhenghuasheng
*/
public class JMSProducer
//默认连接用户名
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
//默认连接密码
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
//默认连接地址
private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
//发送的消息数量
private static final int SENDNUM = 10;
public static void main(String[] args)
//连接工厂
ConnectionFactory connectionFactory;
//连接
Connection connection = null;
//会话 接受或者发送消息的线程
Session session;
//消息的目的地
Destination destination;
//消息生产者
MessageProducer messageProducer;
//实例化连接工厂
connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
try
//通过连接工厂获取连接
connection = connectionFactory.createConnection();
//启动连接
connection.start();
//创建session
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
//创建一个名称为HelloWorld的消息队列
destination = session.createQueue("HelloWorld");
//创建消息生产者
messageProducer = session.createProducer(destination);
//发送消息
sendMessage(session, messageProducer);
session.commit();
catch (Exception e)
e.printStackTrace();
finally
if (connection != null)
try
connection.close();
catch (JMSException e)
e.printStackTrace();
/**
* 发送消息
*
* @param session
* @param messageProducer 消息生产者
* @throws Exception
*/
public static void sendMessage(Session session, MessageProducer messageProducer) throws Exception
for (int i = 0; i < JMSProducer.SENDNUM; i++)
//创建一条文本消息
TextMessage message = session.createTextMessage("ActiveMQ 发送消息" + i);
System.out.println("发送消息：Activemq 发送消息" + i);
//通过消息生产者发出消息
messageProducer.send(message);
运行后 控制台
consumer:
/**
* Created by zhenghuasheng on 2016/8/1.
*/
public class JMSConsumer
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;//默认连接用户名
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址
public static void main(String[] args)
ConnectionFactory connectionFactory;//连接工厂
Connection connection = null;//连接
Session session;//会话 接受或者发送消息的线程
Destination destination;//消息的目的地
MessageConsumer messageConsumer;//消息的消费者
//实例化连接工厂
connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
try
//通过连接工厂获取连接
connection = connectionFactory.createConnection();
//启动连接
connection.start();
//创建session
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建一个连接HelloWorld的消息队列
destination = session.createQueue("HelloWorld");
//创建消息消费者
messageConsumer = session.createConsumer(destination);
while (true)
TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
if(textMessage != null)
System.out.println("收到的消息:" + textMessage.getText());
else
break;
catch (JMSException e)
e.printStackTrace();
运行后 控制台
activeMQ集群配置：https://my.oschina.net/xiaohui249/blog/313028
MQ比较：https://my.oschina.net/javahongxi/blog/1529813
以上是关于ActiveMQ 使用的主要内容，如果未能解决你的问题，请参考以下文章
如何使用虚拟目的地创建多个 activemq 主题订阅者实例？
Spring-integration / ActiveMQ 在单个线程中订阅多个目的地