JMS消息中间件之ActiveMQ学习

1、下载

下载二进制bin文件:http://activemq.apache.org/activemq-5132-release.html

下载源码:

2、启动:

解压任意路径:

启动后:

3、访问:

访问http://localhost:8161/admin/  用户名&密码:admin

4、主要应用:

5、点对点消息发送&接收

首先是producer方:

package com.activemq.test;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息生产者
 * 
 * @author Administrator
 *
 */
public class JMSProducer {

    private static final String USERNAME = ActiveMQConnectionFactory.DEFAULT_USER; // 默认连接用户
    private static final String PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD; // 默认连接密码
    private static final String BROKERURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; // 默认消息总线

    private static final int SENDNUM = 10; // 发送消息总量

    public static void main(String[] args) {

        ConnectionFactory connectionFactory = null; // 连接工厂
        Connection connection = null; // 连接对象

        Session session = null; // 会话级session,接收或发送消息的线程
        Destination destination = null; // 消息发送的目的地

        MessageProducer messageProducer = null; // 消息生产者

        connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD,
                JMSProducer.BROKERURL);
        try {
            connection = connectionFactory.createConnection();

            connection.start(); // 启动连接

            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 创建session,true表示添加事务

            destination = session.createQueue("FirstQueue"); // 创建消息队列

            messageProducer = session.createProducer(destination); // 创建消息生产者
            
            sendMessage(session, messageProducer);
            
            session.commit();            //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 < SENDNUM; i++) {    
            TextMessage textMessage = session.createTextMessage("Active MQ消息"+i);        //文本消息
            System.out.println("发送消息: Active MQ消息"+i);
            messageProducer.send(textMessage);
        }
    }

}

然后是消费方实现,主要有两种,一种是直接receive方法接收消息,一种是通过监听实现:

receive:

package com.activemq.test;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息消费者
 * 普通receive方式接收消息
 * @author Administrator
 *
 */
public class JMSConsumer {
    
    private static final String USERNAME = ActiveMQConnectionFactory.DEFAULT_USER; // 默认连接用户
    private static final String PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD; // 默认连接密码
    private static final String BROKERURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; // 默认消息总线
    public static void main(String[] args) {
        
        ConnectionFactory connectionFactory = null; // 连接工厂
        Connection connection = null; // 连接对象

        Session session = null; // 会话级session,接收或发送消息的线程
        Destination destination = null; // 消息发送的目的地
        
        MessageConsumer messageConsumer = null;        //消息消费者
        
        connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD,
                JMSConsumer.BROKERURL);
        
        try {
            connection = connectionFactory.createConnection();
            
            connection.start(); // 启动连接
            
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 创建session,false表示不添加事务
            
            destination = session.createQueue("FirstQueue"); // 创建消息队列

            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();
        }
        
    }
    
}

监听方式:

package com.activemq.test;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * 消息监听
 * @author Administrator
 *
 */
public class Listener implements MessageListener{

    @Override
    public void onMessage(Message message) {
        
        try {
            System.out.println("接收到的消息:"+((TextMessage)message).getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
        
    }
}
package com.activemq.test;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息消费者
 * 普通receive方式接收消息
 * @author Administrator
 *
 */
public class JMSConsumer2 {
    
    private static final String USERNAME = ActiveMQConnectionFactory.DEFAULT_USER; // 默认连接用户
    private static final String PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD; // 默认连接密码
    private static final String BROKERURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; // 默认消息总线
    public static void main(String[] args) {
        
        ConnectionFactory connectionFactory = null; // 连接工厂
        Connection connection = null; // 连接对象

        Session session = null; // 会话级session,接收或发送消息的线程
        Destination destination = null; // 消息发送的目的地
        
        MessageConsumer messageConsumer = null;        //消息消费者
        
        connectionFactory = new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD,
                JMSConsumer2.BROKERURL);
        
        try {
            connection = connectionFactory.createConnection();
            
            connection.start(); // 启动连接
            
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 创建session,false表示不添加事务
            
            destination = session.createQueue("FirstQueue"); // 创建消息队列

            messageConsumer = session.createConsumer(destination); // 创建消息消费者
            
            messageConsumer.setMessageListener(new Listener());        //注册监听
        } catch (JMSException e) {
            e.printStackTrace();
        }
        
    }
    
}

几轮测试下来,消费生产记录:

6、发布订阅模式

新建订阅1:

package com.activemq.test2;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息消费者    -- 消息订阅者1
 * 普通receive方式接收消息
 * @author Administrator
 *
 */
public class JMSConsumer {
    
    private static final String USERNAME = ActiveMQConnectionFactory.DEFAULT_USER; // 默认连接用户
    private static final String PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD; // 默认连接密码
    private static final String BROKERURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; // 默认消息总线
    public static void main(String[] args) {
        
        ConnectionFactory connectionFactory = null; // 连接工厂
        Connection connection = null; // 连接对象

        Session session = null; // 会话级session,接收或发送消息的线程
        Destination destination = null; // 消息发送的目的地
        
        MessageConsumer messageConsumer = null;        //消息消费者
        
        connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD,
                JMSConsumer.BROKERURL);
        
        try {
            connection = connectionFactory.createConnection();
            
            connection.start(); // 启动连接
            
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 创建session,false表示不添加事务
            
            destination = session.createTopic("SecondTopic"); // 创建消息订阅

            messageConsumer = session.createConsumer(destination); // 创建消息消费者
            
            messageConsumer.setMessageListener(new Listener());        //注册监听
        } catch (JMSException e) {
            e.printStackTrace();
        }
        
    }
    
}
package com.activemq.test2;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * 消息监听-消息订阅者1的监听
 * @author Administrator
 *
 */
public class Listener implements MessageListener{

    @Override
    public void onMessage(Message message) {
        
        try {
            System.out.println("订阅者1接收到的消息:"+((TextMessage)message).getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
        
    }
}

订阅2:

package com.activemq.test2;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息消费者    -- 消息订阅者2
 * 普通receive方式接收消息
 * @author Administrator
 *
 */
public class JMSConsumer2 {
    
    private static final String USERNAME = ActiveMQConnectionFactory.DEFAULT_USER; // 默认连接用户
    private static final String PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD; // 默认连接密码
    private static final String BROKERURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; // 默认消息总线
    public static void main(String[] args) {
        
        ConnectionFactory connectionFactory = null; // 连接工厂
        Connection connection = null; // 连接对象

        Session session = null; // 会话级session,接收或发送消息的线程
        Destination destination = null; // 消息发送的目的地
        
        MessageConsumer messageConsumer = null;        //消息消费者
        
        connectionFactory = new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD,
                JMSConsumer2.BROKERURL);
        
        try {
            connection = connectionFactory.createConnection();
            
            connection.start(); // 启动连接
            
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 创建session,false表示不添加事务
            
            destination = session.createTopic("SecondTopic"); // 创建消息订阅

            messageConsumer = session.createConsumer(destination); // 创建消息消费者
            
            messageConsumer.setMessageListener(new Listener2());        //注册监听
        } catch (JMSException e) {
            e.printStackTrace();
        }
        
    }
    
}
package com.activemq.test2;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * 消息监听-消息订阅者1的监听
 * @author Administrator
 *
 */
public class Listener2 implements MessageListener{

    @Override
    public void onMessage(Message message) {
        
        try {
            System.out.println("订阅者2接收到的消息:"+((TextMessage)message).getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
        
    }
}

消息一定要先订阅,然后producer再发布,否则先发布再订阅的话后边才订阅的一方是收不到之前发布的消息的!

然后是发布方:

package com.activemq.test2;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息生产者
 * 
 * @author Administrator
 *
 */
public class JMSProducer {

    private static final String USERNAME = ActiveMQConnectionFactory.DEFAULT_USER; // 默认连接用户
    private static final String PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD; // 默认连接密码
    private static final String BROKERURL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; // 默认消息总线

    private static final int SENDNUM = 10; // 发送消息总量

    public static void main(String[] args) {

        ConnectionFactory connectionFactory = null; // 连接工厂
        Connection connection = null; // 连接对象

        Session session = null; // 会话级session,接收或发送消息的线程
        Destination destination = null; // 消息发送的目的地

        MessageProducer messageProducer = null; // 消息生产者

        connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD,
                JMSProducer.BROKERURL);
        try {
            connection = connectionFactory.createConnection();

            connection.start(); // 启动连接
            
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 创建session,true表示添加事务

            destination = session.createTopic("SecondTopic"); // 创建发布主题
                
            messageProducer = session.createProducer(destination); // 创建消息发布者
            
            sendMessage(session, messageProducer);

            session.commit(); // 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 < SENDNUM; i++) {
            TextMessage textMessage = session.createTextMessage("Active MQ发布消息" + i); // 文本消息
            System.out.println("发送消息: Active MQ 发布的消息" + i);
            messageProducer.send(textMessage);
        }
    }
}

运行效果查看:

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏杨建荣的学习笔记

linux过滤空文件的命令总结(r2笔记40天)

在linux中有经常做文件的操作,今天有个同事在生产环境统计数据,发现有很多日志文件都是空的,文件太多了,他想查看一下有哪些文件不是空文件。 而且还不想使用脚本...

34270
来自专栏Danny的专栏

【EJB学习笔记】——JMS和消息驱动Bean

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/huyuyang6688/article/...

10820
来自专栏DT乱“码”

activeMQ

activeMQ 是学习java消息队列的实现项目,使用jfinal + jfinal-ext + activeMQ + quartz快速构建。 1.消息队列 ...

266100
来自专栏Java开发者杂谈

ActiveMQ专题2: 持久化

​ 前面一篇AMQ专题中,我们发现对于Topic这种类型的消息,即使将deliveryMode设置为持久化,只要生产者在消费者之前启动。消息生产者发布的消息还是...

10530
来自专栏杨建荣的学习笔记

通过shell和sql结合查找性能sql(r2笔记68天)

在生产系统中,会发现一些潜在的sql问题,为了能够及时和准确的定位,我们可以借助sql_monitor来做性能sql的查找。可以在后台启用一个job不定时的去查...

35570
来自专栏散尽浮华

libmysqlclient.so.16: cannot open shared object file: No such file or directory

编译安装的mysql5.6.39,安装目录是/usr/local/mysql,启用程序时报错:libmysqlclient.so.16: cannot open...

30240
来自专栏帅小子的日常

ActiveMQ消息传递的两种方式

40890
来自专栏个人分享

JMS的常用方法

9810
来自专栏web开发

树结构数据的展示和编辑-zTree树插件的简单使用

最近在项目当中遇到一个需求,需要以树结构的方式展示一些数据,并可对每一个树节点做内容的编辑以及树节点的添加和删除,刚好听说有zTree这个插件可以实现这样的需求...

26610
来自专栏腾讯云容器服务团队的专栏

腾讯云容器服务的滚动升级使用简介

滚动升级是一种多副本服务的升级方式,其特点是能够保证升级过程中服务不中断,对外界无感知。其原理大致为循环的执行以下步奏直至所有容器都被更新,本文将对其使用步骤进...

91500

扫码关注云+社区

领取腾讯云代金券