文档章节

activemq发布订阅

Zero零_度
 Zero零_度
发布于 2016/03/08 17:37
字数 588
阅读 39
收藏 4

关键代码,创建topic

Destination destination = session.createTopic("topic1");

发布者:

package com.sniper.jms.topic;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
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 audaque
 *
 */
public class Sender {
    public static void main(String[] args) {
        ConnectionFactory connectionFactory; // 连接工厂
        Connection connection = null; // 连接
        Session session = null;
        
        try {
            // 实例化连接工厂
            connectionFactory = new ActiveMQConnectionFactory("sniper", "sniper", "tcp://sniper0:61616");
            
            connection = connectionFactory.createConnection(); // 通过连接工厂获取连接
            connection.start(); // 启动连接
            
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session,没有事务
            // 消息的目的地
            Destination destination = session.createTopic("topic1");
            // 创建消息生产者
            MessageProducer messageProducer = session.createProducer(destination); 
            //设置消息不做持久化
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            
            // 发送消息
            for(int i=0; i<2; i++){
                TextMessage message = session.createTextMessage("ActiveMQ 发送的消息"+i);
                System.out.println("发送消息:"+"ActiveMQ 发送的消息"+i);
                messageProducer.send(message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally{
            if(connection!=null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
}

订阅者1:

package com.sniper.jms.topic;

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.ActiveMQConnectionFactory;

/**
 * 订阅者1
 * @author audaque
 *
 */
public class Receiver1 {
    
    public static void main(String[] args) {
        ConnectionFactory connectionFactory; // 连接工厂
        Connection connection = null; // 连接
        Session session = null;
        
        try {
            // 实例化连接工厂
            connectionFactory = new ActiveMQConnectionFactory("sniper", "sniper", "tcp://sniper0:61616");
            connection = connectionFactory.createConnection();  // 通过连接工厂获取连接
            connection.start(); // 启动连接
            
            //自动签收,就是客户端接收到消息之后,会自动给服务端发送消息表示消息已经签收
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
            
            Destination destination = session.createTopic("topic1");
            MessageConsumer messageConsumer = session.createConsumer(destination); // 创建消息消费者
            
            messageConsumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    if(message != null){
                        TextMessage textMessage = (TextMessage)message;
                        try {
                            System.err.println("收到的消息:" + textMessage.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            
            try {
                TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            if(connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
}

订阅者2:

package com.sniper.jms.topic;

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.ActiveMQConnectionFactory;

/**
 * 订阅者2
 * @author audaque
 *
 */
public class Receiver2 {
    
    public static void main(String[] args) {
        ConnectionFactory connectionFactory; // 连接工厂
        Connection connection = null; // 连接
        Session session = null;
        
        try {
            // 实例化连接工厂
            connectionFactory = new ActiveMQConnectionFactory("sniper", "sniper", "tcp://sniper0:61616");
            connection = connectionFactory.createConnection();  // 通过连接工厂获取连接
            connection.start(); // 启动连接
            
            //自动签收,就是客户端接收到消息之后,会自动给服务端发送消息表示消息已经签收
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
            
            Destination destination = session.createTopic("topic1");
            MessageConsumer messageConsumer = session.createConsumer(destination); // 创建消息消费者
            
            messageConsumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    if(message != null){
                        TextMessage textMessage = (TextMessage)message;
                        try {
                            System.err.println("Receiver2收到的消息:" + textMessage.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            
            try {
                TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            if(connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
}


© 著作权归作者所有

共有 人打赏支持
Zero零_度
粉丝 69
博文 1252
码字总数 256097
作品 0
程序员
私信 提问
Android 上有哪些activeMq的实现?

搭建了一个activeMQ服务器,java写了demo订阅/发布这些都没问题。现在在做android上的,就出现问题了。 产生了以下几个问题: 第一个:官方到底有没有提供android客户端版本的实现?还是只要...

wonailmy
2016/09/21
759
1
Linux 安装ActiveMQ(使用Mac远程访问)

阅读本文需要安装JDK 一 ActiveMQ简介 activemq是用java语言编写的一款开源消息总线 activemq是apache出品 activemq消息的传递有两种类型 一种是点对点(即一个生产者和一个消费者一一对应) 另...

梦三
07/15
0
0
消息队列篇—详谈ActiveMQ消息队列模式的分析及使用

消息队列(Message Queue)是分布式系统中重要的组件,通用使用场景可以简单地描述为当不需要立即获得结果,但是并发量需控制时就需要使用消息队列。消息列队有两种消息模式,一种是点对点的...

afreon
04/22
0
0
ActiveMQ专题2: 持久化

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

槟城码农
08/30
0
0
深入浅出 消息队列 ActiveMQ

一、 概述与介绍 ActiveMQ 是Apache出品,最流行的、功能强大的即时通讯和集成模式的开源服务器。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。提供客户端支持跨语言和...

亮liang
2015/05/27
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Ugly Number(leetcode263)

Write a program to check whether a given number is an ugly number. Ugly numbers are positive numbers whose prime factors only include 2, 3, 5. Example 1: Input: 6Output: true......

woshixin
22分钟前
1
0
深度模型从研者 眼里的 似然估计 & Hessain 海森矩阵 & Fisher Information (费雪信息)

深度模型的训练的基本依据是最小化模型拟合数据的误差。旨在不仅知其然(如何构建和训练一个深度模型),还应知其所以然(为什么这样训练,可以做哪些优化)。我们就会发现,有很多研究者,在...

刘小米_思聪
26分钟前
0
0
阿里巴巴智能监控新场景的探索

摘要: 智能监控是智能运维的子领域,详细分析。 作者简介 王肇刚 阿里巴巴全球运行指挥中心高级技术专家 智能监控是智能运维的子领域,我们说的监控,探讨的更多是在监控策略,因为可能从数...

阿里云官方博客
28分钟前
1
0
【双十二】12 月 12 日 12 点 12 分限量抢福利,云计算不剁手!

活动详情 1.首充豪礼 双十二当日,新用户首充 200 元送 200 元代金券!仅限前 88 名!先到先得!拼手速!(新用户指 12 月 12 日前已完成注册但从未充值过的用户)
 2.超值返券 双十二当日,...

七牛云
54分钟前
3
0
Calculate Linux 发布 18 LXQt是基于Gentoo的发行版

导读 Calculate Linux 是一个基于 Gentoo 并与 Gentoo 完全兼容的滚动更新发行版家族,专为企业环境下的快速部署而优化。其提供多个不同的分支版本,分别适用于桌面、服务器等应用场景。该项...

问题终结者
56分钟前
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部