activeMQ 点对点消费者生产者示例
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://10.10.40.174:61616");
Connection connection = null;
Session session = null;
MessageProducer producer = null;
Destination destination = null;
try {
connection = connectionFactory.createConnection("zengjun", "zj");
session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
destination = session.createQueue("TEST.QUEUE.ZJ_02");
producer = session.createProducer(destination);
//设置消息模式,有持久与非持久的
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
connection.start();
double d = Math.random();
InputStream in = new FileInputStream("D:\\2.09M.jar");
BufferedInputStream objBufferedInputStream = new BufferedInputStream(in);
int len = objBufferedInputStream.available();
byte[] bBuffer = new byte[len];
//创建StreamMessage
StreamMessage message = session.createStreamMessage();
objBufferedInputStream.read(bBuffer);
//添加byte数组数据
message.writeBytes(bBuffer);
//添加整型属性
message.setIntProperty("MessageLength1", len);
//添加字符串属性
message.setStringProperty("ID", String.valueOf(d));
in.close();
producer.send(message);
System.out.println("发送消息成功 ID " + String.valueOf(d));
} catch (JMSException e) {
e.printStackTrace();
} catch (Exception ex) {
ex.printStackTrace();
} finally {
//关闭资源
ConnectionUtil.closeSession(session);
ConnectionUtil.closeConnection(connection);
ConnectionUtil.closeMessageProducer(producer);
}
System.out.println("发送结束");
activeMQ 点对点消费者代码示例
// 创建连接工厂
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://10.10.40.174:61616");
Session session = null;
MessageConsumer consumer = null;
Connection connection = null;
Message message = null;
try {
// 访问的用户与密码
connection = connectionFactory.createConnection("ll", "ll");
connection.start();
session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
Destination destination = session.createQueue("TEST.QUEUE.ZJ_02");
consumer = session.createConsumer(destination);
// 无时间参数表示一直等待,直到收到消息。
// message = consumer.receive();
// 有时间参数表示指定时间后没有消息则结束时,如果存在消息就在取完消息后结束
message = consumer.receive(5 * 1000);
// 立即往下执行
// message = consumer.receiveNoWait();
if (message != null) {
System.out.println("收到消息");
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
String text = textMessage.getText();
System.out.println("TEXT:" + text);
textMessage.acknowledge();
} else if (message instanceof StreamMessage) {
StreamMessage streamMessage = (StreamMessage) message;
String strId = streamMessage.getStringProperty("ID");
System.out.println("streammessage ID:" + strId);
streamMessage.acknowledge();
}
} else {
System.out.println("没有收到消息");
}
} catch (Exception e) {
System.out.println("发生异常\n");
e.printStackTrace();
} finally {
//关闭资源
ConnectionUtil.closeAll(connection, session, consumer);
}
activeMQ 发布/订阅,发布者代码示例
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://10.10.40.174:61616");
TopicConnection connection = null;
ActiveMQTopicSession session = null;
ActiveMQTopicPublisher publisher = null;
ActiveMQTopic topic = null;
try {
connection = connectionFactory.createTopicConnection("zengjun", "zj");
session = (ActiveMQTopicSession) connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
topic = (ActiveMQTopic) session.createTopic("TEST.topic.zj");
publisher = (ActiveMQTopicPublisher) session.createPublisher(topic);
publisher.setDeliveryMode(DeliveryMode.PERSISTENT);
connection.start();
int flag = 2;
if (flag == 1) {
TextMessage messageText = session.createTextMessage();
messageText.setText("tipic:" + System.currentTimeMillis());
publisher.publish(messageText);
} else {
StreamMessage messageStream = session.createStreamMessage();
FileInputStream fi = new FileInputStream("D:\\JavaXYQ.zip");
byte[] btyes = new byte[fi.available()];
fi.read(btyes);
messageStream.writeBytes(btyes);
publisher.publish(messageStream);
}
System.out.println("Topic消息发送成功");
} catch (Exception e) {
e.printStackTrace();
} finally {
ConnectionUtil.closeSession(session);
ConnectionUtil.closeConnection(connection);
ConnectionUtil.closeTopicPublisher(publisher);
}
activeMQ 发布/订阅,持久订阅代码示例
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://10.10.40.174:61616");
TopicConnection connection = null;
ActiveMQTopicSession session = null;
ActiveMQTopic topic = null;
ActiveMQTopicSubscriber subscriber = null;
try {
connection = connectionFactory.createTopicConnection("zengjun", "zj");
connection.setClientID("client_ID_test");
connection.start();
session = (ActiveMQTopicSession) connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
topic = (ActiveMQTopic) session.createTopic("TEST.topic.zj");
//创建持久订阅
subscriber = (ActiveMQTopicSubscriber) session.createDurableSubscriber(topic, "Subscriber_name_test");
subscriber.setMessageListener(this);
} catch (Exception e) {
e.printStackTrace();
} finally {
// ConnectionUtil.closeSession(session);
// ConnectionUtil.closeConnection(connection);
// ConnectionUtil.closeTopicSubscriber(subscriber);
}
}
public void onMessage(Message message) {
System.out.println("收到消息");
try {
if (message != null) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
System.out.println(textMessage.getText());
}else if(message instanceof StreamMessage){
System.out.println("收到steam消息");
}
} else {
System.out.println("没有收到消息");
}
Thread.sleep(1 * 1000);
} catch (Exception e) {
e.printStackTrace();
}
}
附件是我在网上搜集的一些资料,以及自己的一些体会文档
分享到:
相关推荐
使用ActiveMQ示例.pdf
activemq示例,里面存在activemq示例的源码,聊天室,实现多人聊天,
本教程旨在帮助activeMQ初学者入门,通过本示例,能完全理解activeMQ的基本概念,为分布式应用打下基础。 本示例中,使用maven管理,完美解决各种依赖问题,不需要自行配置,导入项目等待eclipse自行下载jar包后即可...
分别实现生产者-消费者模式和发布-订阅模式,作为java编程发送消息和消费消息的基础示例。 源码主要包含如下内容: 1.spring boot配置初始化activeMQ 2.队列类型queue,生产者发送队列消息,以及消费者消费相关队列...
activeMQ,java-server java-client,listener监听,留在此处备份
activemq-samples Apache ActiveMQ 示例
ActiveMQ消息示例代码,完整的java项目
自己总结搜集还有自己写的一些ActiveMQ示例。
Spring整合JMS(activeMQ)的示例,开发环境为eclipse+maven
ActiveMQ是一种开源的,实现了JMS规范的,面向消息(MOM)的中间件,为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。
ActiveMQ与Spring整合示例Demo,ActiveMQ安装在linux系统中
这是一个activemq应用的简单示例代码,使用maven搭建的.适合刚刚开始学习activemq的程序员
ActiveMQ是一种开源的,实现了JMS规范的,面向消息(MOM)的中间件,为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。 网盘地址:链接: https://pan.baidu.com/s/1nvOcn0l 密码: 4esh 话说怎么上传...
SpringIntegrationMQ
ActiveMQ RabbitMQ AliyunMQ 示例程序
memcached 和 activeMQ 的入门级示例代码,JAVA eclipse工程
ActiveMQ完整项目示例,用MyEclipse 10开发,欢迎下载学习
Spring Integration整合ActiveMQ示例
其中apache下的ActiveMQ 就是不错的选择。还有一个比较热的是RabbitMQ(是erlang语言实现的)。这里示例下使用 ActiveMQJMS定义了两种方式:Quere(点对点);Topic(发布/订阅)。ConnectionFactory是连接工厂,...