初识activemq消息中间件(p2p模式)

深碍√TFBOYSˉ_ 2022-03-30 08:50 240阅读 0赞

### 初识activemq消息中间件（p2p模式） ###

*  签收模式
 *  事务
 *  持久化
 *  一个生产者一个消费者
 *   *  生产者
     *  消费者
 *  多个生产者和多个消费者
 *   *  生产者
     *  消费者
     *  客户端

# 签收模式 #

> activemq有4中签收模式，分别是CLIENT\_ACKNOWLEDGE、AUTO\_ACKNOWLEDGE、DUPS\_OK\_ACKNOWLEDGE、SESSION\_TRANSACTED

*  CLIENT\_ACKNOWLEDGE，客户端确认签收
 *  AUTO\_ACKNOWLEDGE自动签收，有可能导致消息的丢失,或者消息的重复接受
 *  DUPS\_OK\_ACKNOWLEDGE，"消息可重复"确认，此模式下，可能会出现重复消息，并不是一条消息需要发送多次ACK才行。它是一种潜在的"AUTO\_ACK"确认机制，为批量确认而生，而且具有“延迟”确认的特点。
 *  SESSION\_TRANSACTED，当session使用事务时，就是使用此模式。在事务开启之后，和session.commit()之前，所有消费的消息，要么全部正常确认，要么全部redelivery

# 事务 #

> activemq支持事务，在创建session时指定是否支持事务，如果开启了事务，必须commit。

# 持久化 #

> activemq也可以指定是否持久化消息，可以调用producer.setDeliveryMode(DeliveryMode.NON\_PERSISTENT);来指定是否持久化，默认是持久化的。

# 一个生产者一个消费者 #

> 需要注意的是，生产者和消费者的目的地要一样

## 生产者 ##

public class Producer {
        //用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        //密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        //url
        private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        private static ConnectionFactory factory = null;
        private static Connection connection = null;
        private static Session session = null;
    
    
        public static void main(String[] args) throws JMSException {
            factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, URL);
            connection = factory.createConnection();
            connection.start();
            /*
             *第一个参数：是否打开事务,如果打开了事务，必须commit
             * 第二个参数：消息确认模式，一般设置为CLIENT_ACKNOWLEDGE
             */
            session = connection.createSession(false,Session.SESSION_TRANSACTED);
            //通过Session创建Destination对象，指的是一个客户端用来指定生产消息目标和消费消息来源的对象
            // 在PTP模式中，Destination被称作Queue即队列；
            // 在Pub/Sub模式，Destination被称作Topic即主题。在程序中可以使用多个Queue和Topic。
            Destination queue = session.createQueue("queue");
            //创建一个消息生产者
            MessageProducer producer = session.createProducer(queue);
            //创建一个消息
            TextMessage message = session.createTextMessage("activemq");
            //发送消息
            producer.send(message);
            //session.commit();
            producer.close();
            connection.close();
            session.close();
    
        }
    }

## 消费者 ##

public class Consumer {
        //用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        //密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        //url
        private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        private static Connection connection = null;
        private static Session session = null;
    
        public static void main(String[] args) throws Exception {
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,URL);
            connection = factory.createConnection();
            connection.start();
    
            session = connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
    
            Destination queue = session.createQueue("queue");
    
            MessageConsumer consumer = session.createConsumer(queue);
    
            while(true){
                Thread.sleep(2000);
                TextMessage message = (TextMessage)consumer.receive();
                if(null==message) continue;
                System.out.println(message.getText());
                message.acknowledge();
            }
    
        }
    
    }

# 多个生产者和多个消费者 #

## 生产者 ##

public class Producer implements Runnable {
        //用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        //密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        //url
        private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        private ConnectionFactory factory = null;
        private Connection connection = null;
        private Session session = null;
        private MessageProducer producer1 = null;
        private MessageProducer producer2 = null;
        private MessageProducer producer3 = null;
        private MessageProducer producer4 = null;
        private MessageProducer producer5 = null;
        private int count = 0;
    
        public void run() {
            try {
                factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, URL);
                connection = factory.createConnection();
                connection.start();
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                Destination queue1 = session.createQueue("queue1");
                Destination queue2 = session.createQueue("queue2");
                Destination queue3 = session.createQueue("queue3");
                Destination queue4 = session.createQueue("queue4");
                Destination queue5 = session.createQueue("queue5");
                producer1 = session.createProducer(queue1);
                producer2 = session.createProducer(queue2);
                producer3 = session.createProducer(queue3);
                producer4 = session.createProducer(queue4);
                producer5 = session.createProducer(queue5);
    
                while (count < 100) {
                    count++;
                    //Thread.sleep(1000);
                    int r = this.generateRandom();
                    TextMessage message = session.createTextMessage("activemq消息" + r);
                    switch (r) {
                        case 1:
                            //默认是持久化的
                            producer1.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                            producer1.send(message);
                            break;
                        case 2:
                            producer2.send(message);
                            break;
                        case 3:
                            producer3.send(message);
                            break;
                        case 4:
                            producer4.send(message);
                            break;
                        case 5:
                            producer5.send(message);
                            break;
                        default:
                            System.out.println("错误");
                            break;
                    }
    
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    producer1.close();
                    producer2.close();
                    producer3.close();
                    producer4.close();
                    producer5.close();
                    connection.close();
                    session.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
    
        }
    
        public int generateRandom() {
            Random r = new Random();
            int a = r.nextInt(5) + 1;
            return a;
        }
    
    }

## 消费者 ##

public class Consumer implements Runnable {
        //用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        //密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        //url
        private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        private ConnectionFactory factory = null;
        private Connection connection = null;
        private Session session = null;
        private MessageProducer producer = null;
    
    
        private String name;
        public  Consumer(String name){
            this.name = name;
        }
    
    
        @Override
        public void run() {
           try{
               factory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,URL);
               connection=factory.createConnection();
               connection.start();
               session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
               Destination queue = session.createQueue(name);
               MessageConsumer consumer = session.createConsumer(queue);
               while (true){
                   TextMessage message = (TextMessage) consumer.receive();
                   if(message==null) return;
                   System.out.println(name + "消费了" + message.getText());
                   Thread.sleep(5);
               }
           }catch (Exception e){
               e.printStackTrace();
           }finally {
               try {
                   producer.close();
                   connection.close();
                   session.close();
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
    
        }
    }

## 客户端 ##

public class App {
        public static void main(String[] args) {
            Thread p1 = new Thread(new Producer());
            p1.start();
    
            ExecutorService pool = Executors.newFixedThreadPool(5);
            for (int i = 1; i <= 5; i++) {
                pool.submit(new Consumer("queue" + i));
            }
        }
    }