2. 程式人生 > >[2]AMQP(高階訊息佇列協議) ----QPID不得不說的事

[2]AMQP(高階訊息佇列協議) ----QPID不得不說的事

阿新 • • 發佈：2019-01-02

如果說到AMQP協議，則不得不提的就是QPID。QPID的論壇現在十分的活躍，基本上白天提的問題，晚上馬上就能得到回覆。由此可見QPID的活躍程度。

大家可以到http://qpid.2158936.n2.nabble.com/網站上面註冊一個賬號，真的非常的好，吐血推薦給大家。但是要求大家的因為一定要好。下面的這個例子就是我們當時提的http://qpid.2158936.n2.nabble.com/connect-Qpid-broker-using-the-latest-amqp-1-0-qpid-jms-client-0-9-ssl-td7644406.html

另外，QPID的JIRA系統上面也能報Bug，這個bug就是我們當時報的: https://issues.apache.org/jira/browse/QPIDJMS-183，回覆這麼迅速，作為一個開源專案還是挺給力的。

關於具體的QPID的介紹，請大家參考我上一篇文章，而且安裝方式網路上也有很多，大家可以自己去找。其官方網站為http://qpid.apache.org/ 文件的地址為
http://qpid.apache.org/documentation.html

因為QPID的伺服器支援AMQP-1-0協議，也支援AQMP0-9等協議。所以其最新的客戶端的API的地址為：

（1）qpid-java-6.0.4 支援AMQP 0-10, 0-9-1, 0-9, 0-8

https://qpid.apache.org/releases/qpid-java-6.0.4/

 (2) qpid-jms-0.9 支援AMQP 1-0 協議

初學者很容易混淆上面兩種最新的API的區別.

下面給一個簡單的基於qpid-jms-0.9 客戶端Java API 的傳送和接受訊息的例子。下面的例子我是基於QPID本身提供的Sample的一個改造。借花獻佛，希望對大家有所幫助。

下圖是我專案程式碼的基本結構


(1) hello.properties的檔案配置為

java.naming.factory.initial = org.apache.qpid.jms.jndi.JmsInitialContextFactory
connectionfactory.myFactoryLookup =  amqp://192.168.45.111:5710?jms.username=admin&jms.password=admin&transport.connectTimeout=30000
queue.queue=queue_henry
queue.myQueueLookup=queue_henry

(2) Sender.java的程式碼 
/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */
package org.apache.qpid.jms.example.success;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class Sender {
	private static final String USER = "guest";
	private static final String PASSWORD = "guest";
	private static final int DEFAULT_COUNT = 10;
	private static final int DELIVERY_MODE = DeliveryMode.NON_PERSISTENT;

	private Context getContext() {
		InitialContext context = null;
		try {
			InputStream resourceAsStream = this.getClass().getResourceAsStream("hello.properties");
			Properties properties = new Properties();
			properties.load(resourceAsStream);
			context = new InitialContext(properties);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (NamingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return context;

	}

	public void publish(int count) {
		try {
			// The configuration for the Qpid InitialContextFactory has been
			// supplied in
			// a jndi.properties file in the classpath, which results in it
			// being picked
			// up automatically by the InitialContext constructor.
			Context context = this.getContext();

			ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
			Destination queue = (Destination) context.lookup("myQueueLookup");

			Connection connection = factory.createConnection(USER, PASSWORD);
			connection.setExceptionListener(new MyExceptionListener());
			connection.start();

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

			MessageProducer messageProducer = session.createProducer(queue);

			long start = System.currentTimeMillis();
			for (int i = 1; i <= count; i++) {
				TextMessage message = session.createTextMessage("Text!");
				messageProducer.send(message, DELIVERY_MODE, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);

				if (i % 100 == 0) {
					System.out.println("Sent message " + i);
				}
			}

			long finish = System.currentTimeMillis();
			long taken = finish - start;
			System.out.println("Sent " + count + " messages in " + taken + "ms");

			connection.close();
		} catch (Exception exp) {
			System.out.println("Caught exception, exiting.");
			exp.printStackTrace(System.out);
			System.exit(1);
		}
	}

	public static void main(String[] args) throws Exception {
		int count = DEFAULT_COUNT;
		if (args.length == 0) {
			System.out.println("Sending up to " + count + " messages.");
			System.out
					.println("Specify a message count as the program argument if you wish to send a different amount.");
		} else {
			count = Integer.parseInt(args[0]);
			System.out.println("Sending up to " + count + " messages.");
		}
		Sender sender=new Sender();
		sender.publish(count);

	}

	private static class MyExceptionListener implements ExceptionListener {
		@Override
		public void onException(JMSException exception) {
			System.out.println("Connection ExceptionListener fired, exiting.");
			exception.printStackTrace(System.out);
			System.exit(1);
		}
	}
}

(3) Receiver.java的程式碼

/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */
package org.apache.qpid.jms.example.success;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class Receiver {
	private static final String USER = "guest";
	private static final String PASSWORD = "guest";
	private static final int DEFAULT_COUNT = 10;

	private Context getContext() {
		InitialContext context = null;
		try {
			InputStream resourceAsStream = this.getClass().getResourceAsStream("hello.properties");
			Properties properties = new Properties();
			properties.load(resourceAsStream);
			context = new InitialContext(properties);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (NamingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return context;

	}

	public void receiveMessage(int count) {
		try {
			// The configuration for the Qpid InitialContextFactory has been
			// supplied in
			// a jndi.properties file in the classpath, which results in it
			// being picked
			// up automatically by the InitialContext constructor.

			Context context = this.getContext();//new InitialContext();

			ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
			Destination queue = (Destination) context.lookup("myQueueLookup");

			Connection connection = factory.createConnection(USER, PASSWORD);
			connection.setExceptionListener(new MyExceptionListener());
			connection.start();

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

			MessageConsumer messageConsumer = session.createConsumer(queue);

			long start = System.currentTimeMillis();

			int actualCount = 0;
			boolean deductTimeout = false;
			int timeout = 1000;
			for (int i = 1; i <= count; i++, actualCount++) {
				Message message = messageConsumer.receive(timeout);
				if (message == null) {
					System.out.println("Message " + i + " not received within timeout, stopping.");
					deductTimeout = true;
					break;
				}
				if (i % 100 == 0) {
					System.out.println("Got message " + i);
				}
				System.out.println("The message:"+message.getJMSDestination().toString());
			}

			long finish = System.currentTimeMillis();
			long taken = finish - start;
			if (deductTimeout) {
				taken -= timeout;
			}
			System.out.println("Received " + actualCount + " messages in " + taken + "ms");

			connection.close();
		} catch (Exception exp) {
			System.out.println("Caught exception, exiting.");
			exp.printStackTrace(System.out);
			System.exit(1);
		}
	}

	public static void main(String[] args) throws Exception {
		int count = DEFAULT_COUNT;
		if (args.length == 0) {
			System.out.println("Consuming up to " + count + " messages.");
			System.out.println(
					"Specify a message count as the program argument if you wish to consume a different amount.");
		} else {
			count = Integer.parseInt(args[0]);
			System.out.println("Consuming up to " + count + " messages.");
		}
		Receiver rec=new Receiver();
		rec.receiveMessage(count);
	}

	private static class MyExceptionListener implements ExceptionListener {
		@Override
		public void onException(JMSException exception) {
			System.out.println("Connection ExceptionListener fired, exiting.");
			exception.printStackTrace(System.out);
			System.exit(1);
		}
	}
}


相關推薦

[2]AMQP(高階訊息佇列協議) ----QPID不得不說的事

如果說到AMQP協議，則不得不提的就是QPID。QPID的論壇現在十分的活躍，基本上白天提的問題，晚上馬上就能得到回覆。由此可見QPID的活躍程度。 大家可以到http://qpid.2158936.n2.nabble.com/網站上面註冊一個賬號，真的非常的好，吐血推薦給

RabbitMQ AMQP (高階訊息佇列協議)

目錄 RabbitMQ AMQP (高階訊息佇列協議) AMQP協議是Message Queue訊息佇列的一種協議，RabbitMQ 是基於AMQP協議實現的一種訊息佇列框架。 掌握RabbitMQ，必須要對AMQP的協議有所瞭解，才能使用的得心應手。 本文主要介紹AMQP協議和RabbitMQ的基本概念和

AMQP高階訊息佇列協議

AMQP，即Advanced Message Queuing Protocol,一個提供統一訊息服務的應用層標準高階訊息佇列協議,是應用層協議的一個開放標準,為面向訊息的中介軟體設計。基於此協議的客戶端與訊息中介軟體可傳遞訊息，並不受客戶端/中介軟體不同產品，不同的開發語言等

[1]AMQP(高階訊息佇列協議) ----入門

接觸AMQP協議已經有一段時間了，這個協議比想象的中的要複雜一些，特別是針對AMQP-1-0以前的版本。具體的概念，網上可以搜尋到很多的資料，本文借花獻佛，讓讀者對AMQP協議有一個大概的瞭解。重點是在後續章節的關於一些遇到的技術難點的解決方案的分享。 最新的AMQP協議

深入剖析 RabbitMQ —— Spring 框架下實現 AMQP 高階訊息佇列協議

前言 訊息佇列在現今資料量超大，併發量超高的系統中是十分常用的。本文將會對現時最常用到的幾款訊息佇列框架 ActiveMQ、RabbitMQ、Kafka 進行分析對比。詳細介紹 RabbitMQ 在 Sprinig 框架下的結構及實現原理，從Producer 端的事務、回撥函式（ConfirmCa

常見訊息佇列協議總結

分散式系統中常用通訊模型主要是“請求-應答”模型和“釋出-訂閱”模型。前者常見如RPC通訊，常用HTTP REST或Thrift等協議；後者多指訊息佇列MQ通訊。 RPC大多屬於請求-應答模式，也包括越來越多響應式正規化，對於需要點對點互動、強事務保證和延遲敏感的服務/應用之間的通訊，RPC是優於訊息佇列的。

編譯PHP擴充套件amqp & php訊息佇列 rabbitmq

首先介紹下AMQP: AMQP——高階訊息佇列協議，目前比較有名氣的實現大概就是大名鼎鼎的RabbitMQ了。 RabbitMQ是一個在AMQP基礎上完成的，可複用的企業訊息系統。他遵循Mozilla Public License開源協議。 PHP 下安裝拓展: wget https://github

「訊息佇列」訊息佇列概述與AMQP協議

轉載請註明出處：https://blog.csdn.net/jinixin/article/details/83552185     前面幾篇文章中談了rpc服務, rpc可用於程序間通訊, 使應用得以解耦, 而程序間通訊還可使用訊息佇列來完成. 本篇文章就簡

訊息佇列與AMQP協議

一、 訊息佇列 訊息佇列（Message Queue,簡稱MQ）提供非同步通訊協議，可以實現程序間通訊或同一程序不同執行緒間的通訊。其中‘訊息’是指包含必要資訊的資料。訊息的傳送者傳送完資料後，立即返回，訊息被儲存在訊息隊列當中，對這個訊息感興趣的消費者會訂閱

AMQP訊息佇列的測試方法

作者簡介 羊老師，目前就職於餓了麼物流研發部，運單與服務業務線的測試負責人，同時也負責測試基礎設施的開發與維護，致力於自動化測試及工程效率的提升工作 前言 在大型網際網路架構中經常會用到訊息佇列（Message Queue）這種中介軟體，在服務端測試時，許多測試同學通過工具對API和資料庫都

（五）RabbitMQ訊息佇列-安裝amqp擴充套件並訂閱/釋出Demo（PHP版）

本文將介紹在PHP中如何使用RabbitMQ來實現訊息的訂閱和釋出。我使用的系統依然是Centos7，為了方便，應用伺服器我使用Docker進行部署，容器環境：centos7+nginx+php5.6。 執行環境，安裝AMQP擴充套件： 如何安裝Docker我就不說了，網上很多教程非

XXL-MQ v1.2.2 釋出，分散式訊息佇列

   Release Notes 1、訪問令牌（accessToken）：為提升系統安全性，訊息中心和客戶端進行安全性校驗，雙方AccessToken匹配才允許通訊； 2、支援批量註冊、摘除，提升註冊發現效能；升級 xxl-rpc 至 v1.3.1; 3、升級 pom

PHP高階程式設計之訊息佇列

轉載：https://blog.csdn.net/luyaran/article/details/53034382 1. 什麼是訊息佇列 訊息佇列（英語：Message queue）是一種程序

PHP 高階程式設計之多執行緒-訊息佇列

1. 多執行緒環境安裝 1.1. PHP 5.5.9 安裝PHP 5.5.9 https://github.com/oscm/shell/blob/master/php/5.5.9.sh ./configure --prefix=/srv/php-5.5.9 \ --wi

小白入門微服務(2) - 訊息佇列初體驗

概述 前言 訊息佇列使用場景 什麼是訊息佇列 常用訊息佇列庫對比 Kafka 初體驗 RabbitMQ 初體驗 後記 前言 前面兩篇我們學習了 - 小白入門微服務(0) - 什麼是微服務 - 小白入門微服務(1) - RP

OpenStack訊息佇列AMQP技術具體實現 [分析基於icehouse]

       本文系技術總結，供將來需要時翻開看看， 如有錯誤， 歡迎交流  第一部分：關鍵技術        先發下感慨， 社群的code真是變化太快， code結構發生了很大的重構增強， 關於訊息佇列， openstack.common.rpc慢慢的都會轉移到oslo

php amqp訊息佇列教程1-程式碼實現例項

兩年前曾發過一篇《用Python嘗試RabbitMQ》，沒想到兩年後的今天，基於PHP的amqp資料還是少得可憐，原來的幾個擴充套件也都一個個被廢棄，只剩amqp還健在，且被PECL收錄。雖說被收錄，可官方手冊中的資訊還是略顯單薄。言歸正傳，amqp擴充套件的安裝就不多說了

2.基於redis非同步佇列模組（Reactor模式）-執行緒池還是Redis還是Rabbitmq訊息佇列作為非同步處理的選擇

1.訊息佇列和多執行緒兩者並不衝突，多執行緒可以作為佇列的生產者和消費者。使用外部的訊息佇列時，第一是可以提高應用的穩定性，當程式fail後，寫入外部訊息佇列的資料依舊是儲存的，如果使用兩步commit的佇列的話，可以更加提高這個專案。2. 用執行緒的話，會佔用主伺服器資源，

PHP中使用AMQP訊息佇列

'password'=> 'guest','vhost' =>'/');$conn = new AMQPConnection($conn_args);$conn->connect();//建立exchange名稱和型別$channel = new AMQPChannel($conn);$ex

中介軟體(2)訊息佇列

原文: 中介軟體(2)訊息佇列 　　訊息佇列也是構建大型網站架構過程中非常重要的一箇中間件。 訊息佇列 使用過訊息佇列的同學都知道，訊息佇列最常用的兩個場景是： 1，解耦，一般用在大的網站進行業務拆分的時候，用於各應用之間進行訊息通訊。 2，非同步，針對那些不需要同步執行，可以晚點執行的操作都可以使