2019-07-08 13:24:35 1321浏览
今天千锋扣丁学堂Java培训老师给大家分享一篇关于java如何连接并访问activemq的详细介绍,下面我们一起来看一下吧。
这里还是先放一张图:
<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-core</artifactId> <version>5.7.0</version> </dependency>
package cn.duanjt;
import java.io.IOException;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
/**
* 消费者
* @author 段江涛
* @date 2018-11-23
*/
public class Consumer {
// 全部使用缺省值
private static String USERNAME = ActiveMQConnection.DEFAULT_USER;
private static String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
private static String BROKER = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) throws IOException {
ConnectionFactory factory;
Connection connection = null;
Session session;
Queue queue;
Message message;
MessageConsumer consumer;
try {
factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER);
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, ActiveMQSession.AUTO_ACKNOWLEDGE);
queue = session.createQueue("zd-duanjt");
consumer = session.createConsumer(queue);
//注释部分为同步方式
/*while (true) {
message = consumer.receive();
TextMessage tmsg = (TextMessage) message;
System.out.println("接收到数据:" + tmsg.getText());
}*/
//下面是异步方式,通过注册监听
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage tmsg = (TextMessage) message;
try {
System.out.println("接收到数据:" + tmsg.getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
} catch (Exception e) {
e.printStackTrace();
} finally{
// 之前在这里把连接关闭了,导致一直无法消费数据,注意,这里千万不要关闭连接
}
}
}
package cn.duanjt;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
/**
* 生产者产生数据
* @author 段江涛
* @date 2018-11-23
*/
public class Productor {
// 全部使用缺省值
private static String USERNAME = ActiveMQConnection.DEFAULT_USER;
private static String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
private static String BROKER = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) throws JMSException {
ConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, ActiveMQSession.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("zd-duanjt");
MessageProducer productor = session.createProducer(queue);
for (int i = 0; i < 3; i++) {
String msg = "Hello world:" + i;
Message message = session.createTextMessage(msg);
productor.send(message);
System.out.println("发送数据:" + msg);
}
connection.close();
}
}
【关注微信公众号获取更多学习资料】 【扫码进入JavaEE/微服务VIP免费公开课】