ActiveMQ消息队列详解

activemq消息队列，分为生产者和消费者。

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

1ConnectionFactory connectionFactory;
2
3Connection connection;
4
5Session session;
6
7ThreadLocal<MessageConsumer> threadLocal = new ThreadLocal<>();
8AtomicInteger count \= new AtomicInteger(); public void init(){ try {
9        connectionFactory \= new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN\_URL);
10        connection \= connectionFactory.createConnection();
11        connection.start();
12        session \= connection.createSession(false,Session.AUTO\_ACKNOWLEDGE);
13    } catch (JMSException e) {
14        e.printStackTrace();
15    }
16} public void getMessage(String disname){ try {
17        Queue queue \= session.createQueue(disname);
18        MessageConsumer consumer \= null; if(threadLocal.get()!=null){
19            consumer \= threadLocal.get();
20        }else{
21            consumer \= session.createConsumer(queue);
22            threadLocal.set(consumer);
23        } while(true){
24            Thread.sleep(1000);
25            TextMessage msg \= (TextMessage) consumer.receive(); if(msg!=null) {
26                msg.acknowledge();
27                System.out.println(Thread.currentThread().getName()+": Consumer:我是消费者，我正在消费Msg"+msg.getText()+"--->"+count.getAndIncrement());
28            }else { break;
29            }
30        }
31    } catch (JMSException e) {
32        e.printStackTrace();
33    } catch (InterruptedException e) {
34        e.printStackTrace();
35    }
36}

}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; }

1    @Override public void run() { while(true){ try {
2                producter.sendMessage("Jaycekon-MQ");
3                Thread.sleep(10000);
4            } catch (InterruptedException e) {
5                e.printStackTrace();
6            }
7        }
8    }
9}

}

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; }

1    @Override public void run() { while(true){ try {
2                comsumer.getMessage("Jaycekon-MQ");
3                Thread.sleep(10000);
4            } catch (InterruptedException e) {
5                e.printStackTrace();
6            }
7        }
8    }
9}

}