Java消息队列--ActiveMq 初体验

时间  2019-12-04
标签 java 消息 队列 activemq 体验 栏目 Java 繁體版
原文   原文链接

一、下载安装ActiveMQ

 

　　ActiveMQ官网下载地址：http://activemq.apache.org/download.htmlhtml

　　ActiveMQ 提供了Windows 和Linux、Unix 等几个版本，楼主这里选择了Linux 版本下进行开发。linux

 

 

　　下载完安装包，解压以后的目录：web

　

 　　从它的目录来讲，仍是很简单的： apache

    • bin存放的是脚本文件
    • conf存放的是基本配置文件
    • data存放的是日志文件
    • docs存放的是说明文档
    • examples存放的是简单的实例
    • lib存放的是activemq所需jar包
    • webapps用于存放项目的目录

 

二、启动ActiveMQ　

 　　进入到ActiveMQ 安装目录的Bin 目录，linux 下输入 ./activemq start 启动activeMQ 服务。后端

　　 输入命令以后，会提示咱们建立了一个进程IP 号，这时候说明服务已经成功启动了。浏览器

　　

　　ActiveMQ默认启动时，启动了内置的jetty服务器，提供一个用于监控ActiveMQ的admin应用。 
　　admin：http://127.0.0.1:8161/admin/tomcat

 

　　咱们在浏览器打开连接以后输入帐号密码（这里和tomcat 服务器相似）安全

　　默认帐号：admin服务器

　　密码：adminsession

　　

　　　到这里为止，ActiveMQ 服务端就启动完毕了。

　　　ActiveMQ 在linux 下的终止命令是 ./activemq stop

 

三、建立一个ActiveMQ工程

 

 　　项目目录结构：

　　

　　上述在官网下载ActiveMq 的时候，咱们能够在目录下看到一个jar包：

　　

　　这个jar 包就是咱们须要在项目中进行开发中使用到的相关依赖。

 

　　3.1 建立生产者

复制代码 
public class Producter {

    //ActiveMq 的默认用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    //ActiveMq 的默认登陆密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    //ActiveMQ 的连接地址
    private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    AtomicInteger count = new AtomicInteger(0);
    //连接工厂
    ConnectionFactory connectionFactory;
    //连接对象
    Connection connection;
    //事务管理
    Session session;
    ThreadLocal<MessageProducer> threadLocal = new ThreadLocal<>();

    public void init(){
        try {
            //建立一个连接工厂
            connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
            //从工厂中建立一个连接
            connection  = connectionFactory.createConnection();
            //开启连接
            connection.start();
            //建立一个事务（这里经过参数能够设置事务的级别）
            session = connection.createSession(true,Session.SESSION_TRANSACTED);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void sendMessage(String disname){
        try {
            //建立一个消息队列
            Queue queue = session.createQueue(disname);
            //消息生产者
            MessageProducer messageProducer = null;
            if(threadLocal.get()!=null){
                messageProducer = threadLocal.get();
            }else{
                messageProducer = session.createProducer(queue);
                threadLocal.set(messageProducer);
            }
           while(true){
                Thread.sleep(1000);
                int num = count.getAndIncrement();
                //建立一条消息
                TextMessage msg = session.createTextMessage(Thread.currentThread().getName()+
                        "productor:我是大帅哥，我如今正在生产东西！,count:"+num);
                System.out.println(Thread.currentThread().getName()+
                        "productor:我是大帅哥，我如今正在生产东西！,count:"+num);
                //发送消息
                messageProducer.send(msg);
                //提交事务
                session.commit();
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
复制代码

　　　　 

　　3.2 建立消费者

复制代码 
public class Comsumer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;

    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;

    private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    ConnectionFactory connectionFactory;

    Connection connection;

    Session session;

    ThreadLocal<MessageConsumer> threadLocal = new ThreadLocal<>();
    AtomicInteger count = new AtomicInteger();

    public void init(){
        try {
            connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
            connection  = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }


    public void getMessage(String disname){
        try {
            Queue queue = session.createQueue(disname);
            MessageConsumer consumer = null;

            if(threadLocal.get()!=null){
                consumer = threadLocal.get();
            }else{
                consumer = session.createConsumer(queue);
                threadLocal.set(consumer);
            }
            while(true){
                Thread.sleep(1000);
                TextMessage msg = (TextMessage) consumer.receive();
                if(msg!=null) {
                    msg.acknowledge();
                    System.out.println(Thread.currentThread().getName()+": Consumer:我是消费者，我正在消费Msg"+msg.getText()+"--->"+count.getAndIncrement());
                }else {
                    break;
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
复制代码

 

 

四、运行ActiveMQ项目

 

　　4.1 生产者开始生产消息

复制代码 
public class TestMq {
    public static void main(String[] args){
        Producter producter = new Producter();
        producter.init();
        TestMq testMq = new TestMq();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //Thread 1
        new Thread(testMq.new ProductorMq(producter)).start();
        //Thread 2
        new Thread(testMq.new ProductorMq(producter)).start();
        //Thread 3
        new Thread(testMq.new ProductorMq(producter)).start();
        //Thread 4
        new Thread(testMq.new ProductorMq(producter)).start();
        //Thread 5
        new Thread(testMq.new ProductorMq(producter)).start();
    }

    private class ProductorMq implements Runnable{
        Producter producter;
        public ProductorMq(Producter producter){
            this.producter = producter;
        }

        @Override
        public void run() {
            while(true){
                try {
                    producter.sendMessage("Jaycekon-MQ");
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
复制代码

　　　运行结果：

复制代码 
 INFO | Successfully connected to tcp://localhost:61616
Thread-6productor:我是大帅哥，我如今正在生产东西！,count:0
Thread-4productor:我是大帅哥，我如今正在生产东西！,count:1
Thread-2productor:我是大帅哥，我如今正在生产东西！,count:3
Thread-5productor:我是大帅哥，我如今正在生产东西！,count:2
Thread-3productor:我是大帅哥，我如今正在生产东西！,count:4
Thread-6productor:我是大帅哥，我如今正在生产东西！,count:5
Thread-3productor:我是大帅哥，我如今正在生产东西！,count:6
Thread-5productor:我是大帅哥，我如今正在生产东西！,count:7
Thread-2productor:我是大帅哥，我如今正在生产东西！,count:8
Thread-4productor:我是大帅哥，我如今正在生产东西！,count:9
Thread-6productor:我是大帅哥，我如今正在生产东西！,count:10
Thread-3productor:我是大帅哥，我如今正在生产东西！,count:11
Thread-5productor:我是大帅哥，我如今正在生产东西！,count:12
Thread-2productor:我是大帅哥，我如今正在生产东西！,count:13
Thread-4productor:我是大帅哥，我如今正在生产东西！,count:14
Thread-6productor:我是大帅哥，我如今正在生产东西！,count:15
Thread-3productor:我是大帅哥，我如今正在生产东西！,count:16
Thread-5productor:我是大帅哥，我如今正在生产东西！,count:17
Thread-2productor:我是大帅哥，我如今正在生产东西！,count:18
Thread-4productor:我是大帅哥，我如今正在生产东西！,count:19
复制代码

　　

 

　　4.2 消费者开始消费消息

复制代码 
public class TestConsumer {
    public static void main(String[] args){
        Comsumer comsumer = new Comsumer();
        comsumer.init();
        TestConsumer testConsumer = new TestConsumer();
        new Thread(testConsumer.new ConsumerMq(comsumer)).start();
        new Thread(testConsumer.new ConsumerMq(comsumer)).start();
        new Thread(testConsumer.new ConsumerMq(comsumer)).start();
        new Thread(testConsumer.new ConsumerMq(comsumer)).start();
    }

    private class ConsumerMq implements Runnable{
        Comsumer comsumer;
        public ConsumerMq(Comsumer comsumer){
            this.comsumer = comsumer;
        }

        @Override
        public void run() {
            while(true){
                try {
                    comsumer.getMessage("Jaycekon-MQ");
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
复制代码

　　运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
INFO | Successfully connected to tcp: //localhost:61616
Thread-2: Consumer:我是消费者，我正在消费MsgThread-5productor:我是大帅哥，我如今正在生产东西！,count:4--->0
Thread-3: Consumer:我是消费者，我正在消费MsgThread-4productor:我是大帅哥，我如今正在生产东西！,count:36--->1
Thread-4: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我如今正在生产东西！,count:38--->2
Thread-5: Consumer:我是消费者，我正在消费MsgThread-6productor:我是大帅哥，我如今正在生产东西！,count:37--->3
Thread-2: Consumer:我是消费者，我正在消费MsgThread-6productor:我是大帅哥，我如今正在生产东西！,count:2--->4
Thread-3: Consumer:我是消费者，我正在消费MsgThread-5productor:我是大帅哥，我如今正在生产东西！,count:40--->5
Thread-4: Consumer:我是消费者，我正在消费MsgThread-6productor:我是大帅哥，我如今正在生产东西！,count:42--->6
Thread-5: Consumer:我是消费者，我正在消费MsgThread-4productor:我是大帅哥，我如今正在生产东西！,count:41--->7
Thread-2: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我如今正在生产东西！,count:1--->8
Thread-3: Consumer:我是消费者，我正在消费MsgThread-2productor:我是大帅哥，我如今正在生产东西！,count:44--->9
Thread-4: Consumer:我是消费者，我正在消费MsgThread-4productor:我是大帅哥，我如今正在生产东西！,count:46--->10
Thread-5: Consumer:我是消费者，我正在消费MsgThread-5productor:我是大帅哥，我如今正在生产东西！,count:45--->11
Thread-2: Consumer:我是消费者，我正在消费MsgThread-2productor:我是大帅哥，我如今正在生产东西！,count:3--->12
Thread-3: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我如今正在生产东西！,count:48--->13
Thread-4: Consumer:我是消费者，我正在消费MsgThread-5productor:我是大帅哥，我如今正在生产东西！,count:50--->14
Thread-5: Consumer:我是消费者，我正在消费MsgThread-2productor:我是大帅哥，我如今正在生产东西！,count:49--->15
Thread-4: Consumer:我是消费者，我正在消费MsgThread-2productor:我是大帅哥，我如今正在生产东西！,count:54--->16
Thread-2: Consumer:我是消费者，我正在消费MsgThread-5productor:我是大帅哥，我如今正在生产东西！,count:6--->17
Thread-3: Consumer:我是消费者，我正在消费MsgThread-6productor:我是大帅哥，我如今正在生产东西！,count:52--->18
Thread-5: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我如今正在生产东西！,count:53--->19
Thread-4: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我如今正在生产东西！,count:58--->20

　　

　　查看运行结果，咱们能够作ActiveMQ 服务端：http://127.0.0.1:8161/admin/ 里面的Queues 中查看咱们生产的消息。

 

 

五、ActiveMQ的特性

　5.1 ActiveMq 的特性　

  1. 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
  2. 彻底支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
  3. 对Spring的支持,ActiveMQ能够很容易内嵌到使用Spring的系统里面去,并且也支持Spring2.0的特性
  4. 经过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中经过JCA 1.5 resource adaptors的配置,可让ActiveMQ能够自动的部署到任何兼容J2EE 1.4 商业服务器上
  5. 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
  6. 支持经过JDBC和journal提供高速的消息持久化
  7. 从设计上保证了高性能的集群,客户端-服务器,点对点
  8. 支持Ajax
  9. 支持与Axis的整合
  10. 能够很容易得调用内嵌JMS provider,进行测试

　

　5.2 什么状况下使用ActiveMQ?

    1. 多个项目之间集成 
      (1) 跨平台 
      (2) 多语言 
      (3) 多项目
    2. 下降系统间模块的耦合度，解耦 
      (1) 软件扩展性
    3. 系统先后端隔离 (1) 先后端隔离，屏蔽高安全区文章来自：https://www.cnblogs.com/jaycekon/p/6225058.html
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程