ActiveMQ(二) 转
2024-09-08 10:24:00
package pfs.y2017.m11.mq.activemq.demo02; import java.util.concurrent.atomic.AtomicInteger; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; 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();
}
}
}
package pfs.y2017.m11.mq.activemq.demo02; import java.util.concurrent.atomic.AtomicInteger; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; 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();
}
}
}
package pfs.y2017.m11.mq.activemq.demo02;
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-MQ2");
// Thread.sleep(10000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
package pfs.y2017.m11.mq.activemq.demo02;
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-MQ2");
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
最新文章
- java网络编程精解demo1---读取用户控制台的输入的数据并显示
- Linux服务器init 5启动图形界面，报错Retrigger failed udev events的解决方法
- [转载]Matlab之静态文本多行输出
- C# ref的应用
- VS附加到进程调试的方法及应用场景
- iOS开发——UI篇OC篇&TextField作为搜索框的使用
- mysql运算符的优先级
- sencha app build 到 Capturing theme image不执行
- 北京创客空间 BEIJING MAXPACE的小站
- 阿里2015回顾面试招收学历（获得成功offer）
- win10锁屏壁纸路径
- 转：【Java集合源码剖析】LinkedHashmap源码剖析
- 201521123020《java程序设计》 第11周学习总结
- C++11 作用域内枚举
- 开发Oracle 函数
- git 解决授权失败的方法
- [20181130]control file sequential read.txt
- 【Sql】经典sql语句
- phoneGap使用 (MAC)
- Mac系统安装MyEclipse