Java中間消息件ActiveMQ使用實(shí)例

更新時(shí)間：2019年11月14日 08:58:07 作者：小魚吃貓

這篇文章主要介紹了Java中間消息件ActiveMQ使用實(shí)例,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

先來說一說我們?yōu)槭裁匆眠@個(gè)東西??！

比如，我們現(xiàn)在有這樣了個(gè)問題要解決：

這樣，我們就要用到中間消息間了

然后我們就說一下什么是中間消息間吧。

采用消息傳送機(jī)制/消息隊(duì)列的中間件技術(shù)，進(jìn)行數(shù)據(jù)交流，用在分布式系統(tǒng)的集成。
Java中對(duì)Jms有了定義，這是Java消息的統(tǒng)一接口。什么是ActiveMq呢？這是這個(gè)接口的一種實(shí)現(xiàn)，相當(dāng)于數(shù)據(jù)庫連接驅(qū)動(dòng)一樣，不同廠商有自己不同的實(shí)現(xiàn)，我們盡快看怎么用代碼實(shí)現(xiàn)吧。

消息一共有兩種接收和發(fā)送形式：點(diǎn)對(duì)點(diǎn)和發(fā)布定閱模式，也就是“一對(duì)一”和“一對(duì)多”。

1.導(dǎo)包（maven)：

  <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-client</artifactId>
      <version>5.13.4</version>
    </dependency>

2.開始寫類，提供者(發(fā)送者)和消費(fèi)者(接收者)是兩個(gè)不同的項(xiàng)目,我們先創(chuàng)建普通的maven項(xiàng)目，而不是web項(xiàng)目點(diǎn)對(duì)點(diǎn)的方式(消息只能被消費(fèi)一次，如果同時(shí)有多個(gè)消費(fèi)者，誰先搶到就是誰的)

消息提供者

 public static void main(String[] args) throws JMSException {

    //創(chuàng)建連接工廠，這個(gè)參數(shù)就是自己的activeMQ的地址
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.180:61616");

    //2.創(chuàng)建連接
    Connection connection = connectionFactory.createConnection();

    //3.啟動(dòng)連接
    connection.start();

    //4.獲取session(會(huì)話對(duì)象)
    /*
    arg0 是否啟用事務(wù)
    arg1 消息的確認(rèn)方式 自動(dòng)確認(rèn)
     */
    Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

    //5.創(chuàng)建一個(gè)隊(duì)列對(duì)象,名稱
    Queue firstQueue = session.createQueue("firstQueue");

    //6.創(chuàng)建一個(gè)消息的生產(chǎn)者對(duì)象
//    Destination destination = ;//目標(biāo)對(duì)象
    MessageProducer producer = session.createProducer(firstQueue);

    //7.創(chuàng)建一個(gè)消息
    TextMessage textMessage = session.createTextMessage("歡迎來到奇的天喻軟件");

    //8.發(fā)送消息
    producer.send(textMessage);
    //9.關(guān)閉資源
    producer.close();
    session.close();
    connection.close();

  }

消息消費(fèi)者

前幾步是一樣的，都是創(chuàng)建連接，只有第6步不一樣,創(chuàng)建的是一個(gè)消費(fèi)者

public static void main(String[] args) throws JMSException, IOException {
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.180:61616");

    //2.創(chuàng)建連接
    Connection connection = connectionFactory.createConnection();

    //3.啟動(dòng)連接
    connection.start();

    //4.獲取session(會(huì)話對(duì)象)
    /*
    arg0 是否啟用事務(wù)
    arg1 消息的確認(rèn)方式 自動(dòng)確認(rèn)
     */
    Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

    //5.創(chuàng)建一個(gè)隊(duì)列對(duì)象,名稱
    Queue firstQueue = session.createQueue("firstQueue");



    //6.創(chuàng)建消息消費(fèi)者對(duì)象
    MessageConsumer consumer = session.createConsumer(firstQueue);

    //7.設(shè)置監(jiān)聽
    consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage)message;
        try {
          System.out.println("提取的消息是"+textMessage.getText());
        } catch (JMSException e) {
          e.printStackTrace();
        }
      }
    });

    //8.等待鍵盤輸入
    //目的是為了讓程序停止來看效果
    System.in.read();


    //9.關(guān)閉資源
    consumer.close();
    session.close();
    connection.close();
  }

發(fā)布訂閱模式（發(fā)布消息后，只有在之前運(yùn)行的消費(fèi)者才能收到，消息被任何一個(gè)消費(fèi)者消費(fèi)后，以后啟動(dòng)的消費(fèi)者不能消費(fèi)之前的消息）

消息提供者

 //創(chuàng)建連接工廠
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.180:61616");

    //2.創(chuàng)建連接
    Connection connection = connectionFactory.createConnection();

    //3.啟動(dòng)連接
    connection.start();

    //4.獲取session(會(huì)話對(duì)象)
    /*
    arg0 是否啟用事務(wù)
    arg1 消息的確認(rèn)方式 自動(dòng)確認(rèn)
     */
    Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

    //5
    Topic topic = session.createTopic("first-topic");


    //6.創(chuàng)建一個(gè)消息的生產(chǎn)者對(duì)象
//    Destination destination = ;//目標(biāo)對(duì)象
    MessageProducer producer = session.createProducer(topic);

    //7.創(chuàng)建一個(gè)消息
    TextMessage textMessage = session.createTextMessage("歡迎來到奇的天喻軟件");

    //8.發(fā)送消息
    producer.send(textMessage);


    //9.關(guān)閉資源
    producer.close();
    session.close();
    connection.close();

消費(fèi)者

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

    //2.創(chuàng)建連接
    Connection connection = connectionFactory.createConnection();

    //3.啟動(dòng)連接
    connection.start();

    //4.獲取session(會(huì)話對(duì)象)
    /*
    arg0 是否啟用事務(wù)
    arg1 消息的確認(rèn)方式 自動(dòng)確認(rèn)
     */
    Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

    //5
    Topic topic = session.createTopic("first-topic");



    //6.創(chuàng)建消息消費(fèi)者對(duì)象
    MessageConsumer consumer = session.createConsumer(topic);

    //7.設(shè)置監(jiān)聽
    consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage)message;
        try {
          System.out.println("提取的消息是"+textMessage.getText());
        } catch (JMSException e) {
          e.printStackTrace();
        }
      }
    });
    //8.等待鍵盤輸入
    //目的是為了讓程序停止來看效果
    System.in.read();
    //9.關(guān)閉資源
    consumer.close();
    session.close();
    connection.close();

總結(jié)，是不是發(fā)現(xiàn)上邊代碼都很相似，那么完全可以用Spring來管理了啊

以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: