Java实现

 

添加相应的jar包

<dependency>

  <groupId>org.apache.activemq</groupId>

  <artifactId>activemq-all</artifactId>

  <version>x.xx.x</version>

</dependency>

 

创建生产者类（点对点）

public class ProducerTest {

    // 异步发送asyn

    // 死信队列 DLQ

    // 文件上传

    // header properties使用

    // jdbc存储

    // byteMsg objMsg inputMsg

    // mq ptp 的使用场景

    // mq中所有的队列名以及每个队列中未被消费的消息数量

    // mq sub/pub 的使用场景

    @Test

    public void testQueueProducer() throws JMSException{

       // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。

       //brokerURL服务器的ip及端口号

       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.114.129:61616");

       // 第二步：使用ConnectionFactory对象创建一个Connection对象

       Connection connection = connectionFactory.createConnection();

       // 第三步：开启连接，调用Connection对象的start方法

       connection.start();

       // 第四步：使用Connection对象创建一个Session对象。

       //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。

       //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答

       Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

       // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。

       //参数：队列的名称。

       Queue queue = session.createQueue("test-queue");

       // 第六步：使用Session对象创建一个Producer对象。

       MessageProducer producer = session.createProducer(queue);

       producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

       // 第七步：创建一个Message对象，创建一个TextMessage对象。

       /*TextMessage message = new ActiveMQTextMessage();

       message.setText("hello activeMq,this is my first test.");*/

       TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");

       producer.send(textMessage);

       producer.close();

       session.close();

       connection.close();

    }

}

 

创建消费者类（点对点）

public class ConsumerTest {

    @Test

    public void testQueueConsumer() throws JMSException, IOException{

       // 第一步：创建一个ConnectionFactory对象。

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.114.129:61616");

        // 第二步：从ConnectionFactory对象中获得一个Connection对象。

        Connection connection = connectionFactory.createConnection();

        // 第三步：开启连接。调用Connection对象的start方法。

        connection.start();

        // 第四步：使用Connection对象创建一个Session对象。

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。

        Queue queue = session.createQueue("queue.stu");

        // 第六步：使用Session对象创建一个Consumer对象。

        MessageConsumer consumer = session.createConsumer(queue);

        // 第七步：接收消息。

        consumer.setMessageListener(new MessageListener()

             @Override

             public void onMessage(Message message) {

                  try {

                    //if (TextMessage.class.isAssignableFrom(message.getClass()))

                    if (message instanceof TextMessage) {

                          TextMessage textMessage = (TextMessage) message;

                          String text = null;

                          //取消息的内容

                          text = textMessage.getText();

                          // 第八步：打印消息。

                          System.out.println(text);

                    }

                  } catch (JMSException e) {

                      e.printStackTrace();

                  }

             }

        });

        //等待键盘输入

        System.in.read();

        // 第九步：关闭资源

        consumer.close();

        session.close();

        connection.close();

    }

}

 

运行active服务器验证实现情况

 

发布（对点对点代码修改 创建Topic Destination）

// 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个topic对象。

// 参数：话题的名称。

Topic topic = session.createTopic("test-topic");

// 第六步：使用Session对象创建一个Producer对象。

MessageProducer producer = session.createProducer(topic);

 

订阅（对点对点代码修改 创建Topic Destination）

// 第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。

Topic topic = session.createTopic("test-topic");

// 第六步：使用Session对象创建一个Consumer对象。

MessageConsumer consumer = session.createConsumer(topic);

 

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有