ActiveMQ专题1: 入门实例

好久没有写博客了,最近真的是可以说是忙成狗了。项目的事和自己的终身大事忙得焦头烂额,好在是一切都是越来越好了......
趁着项目今天唯一的一点喘息时间,加上项目开始接触到的mq,开始写一篇amq的入门专提

AMQ入门实例

下载导入源码:

  • 下载地址: http://activemq.apache.org/activemq-5155-release.html (可以同时下载安装包和源码,我这里为了和项目中使用的保持一致,下载的是5.8.0版本)
  • 源码导入方式: 先通过mvn编译,然后导入

管理后台

  • 通过bin目录下的activemq.bat启动之后,就可以通过: http://localhost:8161/admin 来访问activemq的管理后台了
  • 默认的用户名和密码都是: admin. 用户名和密码配置在conf目录下的jetty-realm.properties文件中
  • 访问的端口配置在conf目录下的jetty.xml文件中

入门使用实例

1. 引入mq依赖

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.8.0</version>
</dependency>

2. 启动activemq服务

3. 简单的服务端实现

public class SimpleProducer {
    public static void main(String[] args) {
        // STEP1: 得到连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
        
        Connection connection = null;
        Session session = null;
        Destination destination = null;
        MessageProducer producer = null;
        MessageProducer topicProducer = null;
        Destination topicDestination = null;
        try {
            // STEP2: 从连接工厂得到连接并且启动连接
            connection = connectionFactory.createConnection();
            connection.start();
            
            // STEP3: 获取会话
            /**
             * 第一个参数表示是否开启事务:
             * 当第一个参数为true的时候,会忽略第二个参数,无论第二个参数为啥,都需要显示调用 session.commit() 消息才会提交到MQ
             * 当第一个参数为false的时候,第二个参数不能为:Session.SESSION_TRANSACTED。 且当第二个参数为其他合法值时,都不需要调用 session.commit(),消息都会发送到MQ
             * 第二个参数表示当未开启事务的时候,消费者或者客户端在什么时候发送确认消息
             */
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            
            // STEP4: 创建目标队列、主题 
            /**
             * 队列和主题的区别在于:
             * 1、 队列是点对点的,队列中的消息只会被消费一次
             * 2、 主题类似于广播机制,只要订阅了该主题的消费者都可以对该消息进行消费
             * 3、 一般来说如果生产者在消费者启动之前创建了主题,那么消费者启动后接收不到主题。
             */
            destination = session.createQueue("KiDe-Demo");
            topicDestination = session.createTopic("KiDe-Demo");
            
            // STEP5: 创建消息生产者
            producer = session.createProducer(destination);
            topicProducer = session.createProducer(topicDestination);
            
            /**
             * 参数表示生产者发送的消息是否进行持久化
             */
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);      // 设置不持久化
            topicProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);     // 设置不持久化
            
            // STEP6: 发送消息
            for (int i=0; i<20; i++) {
                TextMessage message = session.createTextMessage("Producer message:" + i);
                producer.send(message);
                topicProducer.send(message);
            }
            
            // STEP7: 如果开启了事务 ,此时需要调用session提交操作
            // session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                }
            }
        }
    }
}

简单消费者实现

package com.rampage.learning.activemq;

import java.util.concurrent.TimeUnit;

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

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 简单的队列消费者
 * 
 * @author ziyuqi
 *
 */
public class SimpleConsumer {
    public static void main(String[] args) {
        // STEP1: 创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);

        Connection connection = null;
        Session session = null;
        Destination destination = null;
        Destination topicDestination = null;
        MessageConsumer consumer = null;
        MessageConsumer topicConsumer = null;
        try {
            // STEP2: 从连接工厂得到连接并且启动连接
            connection = connectionFactory.createConnection();
            connection.start();

            // STEP3: 获取会话
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // STEP4: 创建目标队列
            destination = session.createQueue("KiDe-Demo");
            topicDestination = session.createTopic("KiDe-Demo");
            
            // STEP5: 创建消费者
            consumer = session.createConsumer(destination);
            topicConsumer = session.createConsumer(topicDestination);
            
            // STEP6: 设置消息接收者接收消息 也可以通过死循环接收消息
            /*while (true) {
                TextMessage textMessage = (TextMessage) consumer.receive(1000);
                System.out.println(textMessage.getText());
            }*/
            consumer.setMessageListener(new MessageListener() {
                
                @Override
                public void onMessage(Message paramMessage) {
                    TextMessage message = (TextMessage) paramMessage;
                    try {
                        System.out.println("消费者接收到队列消息:" + message.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            topicConsumer.setMessageListener(new MessageListener() {
                
                @Override
                public void onMessage(Message paramMessage) {
                    TextMessage message = (TextMessage) paramMessage;
                    try {
                        System.out.println("消费者接收到主题消息:" + message.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            TimeUnit.SECONDS.sleep(200);    // 睡眠20秒,使得客户端可以接收到对应消息
        } catch (Exception e) {

        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                }
            }
        }
    }
}

代码说明

​ 从上面的代码可以看出,生产者和消费者的处理流程大致相同。存在很多重复代码,不难发现可以抽取出公共的代码来使得代码更加简洁。

运行结果说明

我这里运行了producer后,运行了两个consumer。不难发现,topic中的每条消息会被每个consumer完全消费,而queue中的消息,每一条消息只会被两个consumer中的一个消费。

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏测试开发架构之路

centos7安装python3.6后导致防火墙功能无法正常工作的解决办法

问题:因为默认python版本被设置成了python3.6,而进行防火墙的指令操作频频报错。

9910
来自专栏Java开发者杂谈

ActiveMQ专题2: 持久化

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

11330
来自专栏编程

ActiveMQ消息队列的使用及应用

ActiveMQ消息队列的使用及应用 这里就不说怎么安装了,直接解压出来就行了。 谢绝转载,作者保留所有权力 目录: 复制代码 一:JMQ的两种消息模式 1.1...

35370
来自专栏菩提树下的杨过

ActiveMQ笔记(1):编译、安装、示例代码

一、编译 虽然ActiveMQ提供了发布版本,但是建议同学们自己下载源代码编译,以后万一有坑,还可以尝试自己改改源码。 1.1 https://github.c...

32250
来自专栏老安的博客

原创:用zabbix api批量添加web监控

28340
来自专栏北京马哥教育

如何快速截取某段时间内的日志

在排除故障时,需要分析发生故障的原因,避免再次出现同样的问题,需要对日志进行分析,截取故障前后时间段的日志。通常用正则表达式来获取某段时间的内的日志,比较繁琐。...

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

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

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

92200
来自专栏散尽浮华

线上图片批量更换脚本记录

需求说明: 线上之前使用的图片没有经过处理,比较大。应产品同事要求,将图片拿出来处理后再替换之前的图片,图片名不变。 /tmp/image目录下存放的修改后的图...

22060
来自专栏禁心尽力

这次真的忽略了一些ActiveMQ内心的娇艳

首先ActiveMQ有俩种消息队列模式:点对点和发布订阅,这俩种都有不可替代的应用场景,前者适用于消息唯一传递的业务,后者适用于分布式环境下进行多面数据同步的...

35790
来自专栏积累沉淀

干货--JMS(java消息服务)整合Spring项目案例

Sprng-jms消息服务小项目 所需的包: spring的基础包 spring-jms-xx包 spring-messag...

565100

扫码关注云+社区

领取腾讯云代金券