2. 程式人生 > >PHP高階程式設計之訊息佇列

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

阿新 • • 發佈：2018-12-22

轉載：https://blog.csdn.net/luyaran/article/details/53034382

1. 什麼是訊息佇列

訊息佇列（英語：Message queue）是一種程序間通訊或同一程序的不同執行緒間的通訊方式

2. 為什麼使用訊息佇列

訊息佇列技術是分散式應用間交換資訊的一種技術。訊息佇列可駐留在記憶體或磁碟上,佇列儲存訊息直到它們被應用程式讀出。通過訊息佇列，應用程式可獨立地執行，它們不需要知道彼此的位置、或在繼續執行前不需要等待接收程式接收此訊息。

3. 什麼場合使用訊息佇列

你首先需要弄清楚，訊息佇列與遠端過程呼叫的區別，在很多讀者諮詢我的時候，我發現他們需要的是RPC(遠端過程呼叫)，而不是訊息佇列。

訊息佇列有同步或非同步實現方式，通常我們採用非同步方式使用訊息佇列，遠端過程呼叫多采用同步方式。

MQ與RPC有什麼不同？ MQ通常傳遞無規則協議，這個協議由使用者定義並且實現儲存轉發；而RPC通常是專用協議，呼叫過程返回結果。

4. 什麼時候使用訊息佇列

同步需求，遠端過程呼叫(PRC)更適合你。

非同步需求，訊息佇列更適合你。

目前很多訊息佇列軟體同時支援RPC功能，很多RPC系統也能非同步呼叫。

訊息佇列用來實現下列需求

  1. 儲存轉發

  2. 分散式事務

  3. 釋出訂閱

  4. 基於內容的路由

  5. 點對點連線

5. 誰負責處理訊息佇列

通常的做法，如果小的專案團隊可以有一個人實現，包括訊息的推送，接收處理。如果大型團隊，通常是定義好訊息協議，然後各自開發各自的部分，例如一個團隊負責寫推送協議部分，另一個團隊負責寫接收與處理部分。

那麼為什麼我們不講訊息佇列框架化呢？

框架化有幾個好處：
  1. 開發者不用學習訊息佇列介面
  2. 開發者不需要關心訊息推送與接收
  3. 開發者通過統一的API推送訊息
  4. 開發者的重點是實現業務邏輯功能

6. 怎麼實現訊息佇列框架

下面是作者開發的一個SOA框架，該框架提供了三種介面，分別是SOAP,RESTful,AMQP(RabbitMQ)，理解了該框架思想，你很容易進一步擴充套件，例如增加XML-RPC, ZeroMQ等等支援。

本文只講訊息佇列框架部分。

6.1. 守護程序

訊息佇列框架是本地應用程式（命令列程式），我們為了讓他在後臺執行，需要實現守護程序。

每個例項處理一組佇列，例項化需要提供三個引數，$queueName = '佇列名', $exchangeName = '交換名', $routeKey = '路由'

$daemon = new \framework\RabbitDaemon($queueName = 'email', $exchangeName = 'email', $routeKey = 'email');

守護程序需要使用root使用者執行，執行後會切換到普通使用者，同時建立程序ID檔案，以便程序停止的時候使用。

6.2. 訊息佇列協議

訊息協議是一個數組，將陣列序列化或者轉為JSON推送到訊息佇列伺服器，這裡使用json格式的協議。

$msg = array(
	'Namespace'=>'namespace',
	"Class"=>"Email",
	"Method"=>"smtp",
	"Param" => array(
		$mail, $subject, $message, null
	)
);

序列化後的協議

{"Namespace":"single","Class":"Email","Method":"smtp","Param":["[email protected]","Hello"," TestHelloWorld",null]}

使用json格式是考慮到通用性，這樣推送端可以使用任何語言。如果不考慮相容，建議使用二進位制序列化，例如msgpack效率更好。

6.3. 訊息佇列處理

訊息佇列處理核心程式碼

所以訊息的處理在下面一段程式碼中進行

$this->queue->consume(function($envelope, $queue) {

$speed = microtime(true);

$msg = $envelope->getBody();
$result = $this->loader($msg);
$queue->ack($envelope->getDeliveryTag()); //手動傳送ACK應答

//$this->logging->info(''.$msg.' '.$result)
$this->logging->debug('Protocol: '.$msg.' ');
$this->logging->debug('Result: '. $result.' ');
$this->logging->debug('Time: '. (microtime(true) - $speed) .'');

});

public function loader($msg = null) 負責拆解協議，然後載入對應的類檔案，傳遞引數，執行方法，反饋結果。

Time 可以輸出程式執行所花費的時間，對於後期優化十分有用。

提示

loader() 可以進一步優化，使用多執行緒每次呼叫loader將任務提交到執行緒池中，這樣便可以多執行緒處理訊息佇列。

6.4. 測試

			
<?php
$queueName = 'example';
$exchangeName = 'email';
$routeKey = 'email';
$mail = $argv[1];
$subject = $argv[2];
$message = empty($argv[3]) ? 'Hello World!' : ' '.$argv[3];

$connection = new AMQPConnection(array(
‘host’ => ‘192.168.4.1’,
‘port’ => ‘5672’,
‘vhost’ => ‘/’,
‘login’ => ‘guest’,
‘password’ => ‘guest’
));
$connection->connect() or die(“Cannot connect to the broker!\n”);

channel=newAMQPChannel(channel = new AMQPChannel(connection);
exchange=newAMQPExchange(exchange = new AMQPExchange(channel);
KaTeX parse error: Expected 'EOF', got '&' at position 10: exchange-&̲gt;setName(exchangeName);
queue=newAMQPQueue(queue = new AMQPQueue(channel);
KaTeX parse error: Expected 'EOF', got '&' at position 7: queue-&̲gt;setName(queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();

$msg = array(
‘Namespace’=>‘namespace’,
“Class”=>“Email”,
“Method”=>“smtp”,
“Param” => array(
$mail, $subject, $message, null
)
);

KaTeX parse error: Expected 'EOF', got '&' at position 10: exchange-&̲gt;publish(json…msg), routeKey);printf(&quot;[x]SentrouteKey); printf(&quot;[x] Sent %s \r\n&quot;, json_encode(msg));

$connection->disconnect();

		</span></strong></pre>

這裡只給出了少量測試與演示程式，如有疑問請到瀆者群，或者公眾號詢問。

7. 多執行緒

上面訊息佇列 核心程式碼如下

		
$this->queue->consume(function($envelope, $queue) {	
	$msg = $envelope->getBody();
	$result = $this->loader($msg);
	$queue->ack($envelope->getDeliveryTag());
});

	</span></strong></pre>

這段程式碼生產環境使用了半年，發現效率比較低。有些業務場入隊非常快，但處理起來所花的時間就比較長，容易出現佇列堆積現象。

增加多執行緒可能更有效利用硬體資源，提高業務處理能力。程式碼如下

		
<?php
namespace framework;

require_once( DIR.’/autoload.class.php’ );

class RabbitThread extends \Threaded {

private $queue;
public $classspath;
protected $msg;

public function __construct($queue, $logging, $msg) {
	$this-&gt;classspath = __DIR__.'/../queue';
	$this-&gt;msg = $msg;
	$this-&gt;logging = $logging;
	$this-&gt;queue = $queue;
}
public function run() {
	$speed = microtime(true);
	$result = $this-&gt;loader($this-&gt;msg);
	$this-&gt;logging-&gt;debug('Result: '. $result.' ');
	$this-&gt;logging-&gt;debug('Time: '. (microtime(true) - $speed) .'');
}
// private
public  function loader($msg = null){
	
	$protocol 	= json_decode($msg,true);
	$namespace	= $protocol['Namespace'];
	$class 		= $protocol['Class'];
	$method 	= $protocol['Method'];
	$param 		= $protocol['Param'];
	$result 	= null;

	$classspath = $this-&gt;classspath.'/'.$this-&gt;queue.'/'.$namespace.'/'.strtolower($class)  . '.class.php';
	if( is_file($classspath) ){
		require_once($classspath);
		//$class = ucfirst(substr($request_uri, strrpos($request_uri, '/')+1));
		if (class_exists($class)) {
			if(method_exists($class, $method)){
				$obj = new $class;
				if (!$param){
					$tmp = $obj-&gt;$method();
					$result = json_encode($tmp);
					$this-&gt;logging-&gt;info($class.'-&gt;'.$method.'()');
				}else{
					$tmp = call_user_func_array(array($obj, $method), $param);
					$result = (json_encode($tmp));
					$this-&gt;logging-&gt;info($class.'-&gt;'.$method.'("'.implode('","', $param).'")');
				}
			}else{
				$this-&gt;logging-&gt;error('Object '. $class. '-&gt;' . $method. ' is not exist.');
			}
		}else{
			$msg = sprintf("Object is not exist. (%s)", $class);
			$this-&gt;logging-&gt;error($msg);
		}
	}else{
		$msg = sprintf("Cannot loading interface! (%s)", $classspath);
		$this-&gt;logging-&gt;error($msg);
	}
	return $result;
}

}

class RabbitMQ {

const loop = 10;

protected $queue;
protected $pool;

public function __construct($queueName = '', $exchangeName = '', $routeKey = '') {

	$this-&gt;config = new \framework\Config('rabbitmq.ini');
	$this-&gt;logfile = __DIR__.'/../log/rabbitmq.%s.log';
	$this-&gt;logqueue = __DIR__.'/../log/queue.%s.log';
	$this-&gt;logging = new \framework\log\Logging($this-&gt;logfile, $debug=true);

 //.H:i:s

	$this-&gt;queueName	= $queueName;
	$this-&gt;exchangeName	= $exchangeName;
	$this-&gt;routeKey		= $routeKey; 

	$this-&gt;pool = new \Pool($this-&gt;config-&gt;get('pool')['thread']);

}
public function main(){
	
	$connection = new \AMQPConnection($this-&gt;config-&gt;get('rabbitmq'));
	try {
		$connection-&gt;connect();
		if (!$connection-&gt;isConnected()) {
			$this-&gt;logging-&gt;exception("Cannot connect to the broker!"

.PHP_EOL);
}
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;channel = ne…connection);
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;exchange = n…this->channel);
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;exchange-&gt…this->exchangeName);
$this->exchange->setType(AMQP_EX_TYPE_DIRECT); //direct型別
$this->exchange->setFlags(AMQP_DURABLE); //持久�?
$this->exchange->declareExchange();
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;queue = new …this->channel);
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;queue-&gt;se…this->queueName);
$this->queue->setFlags(AMQP_DURABLE); //持久�?
$this->queue->declareQueue();

		$this-&gt;queue-&gt;bind($this-&gt;exchangeName, $this-&gt;routeKey);

		$this-&gt;queue-&gt;consume(function($envelope, $queue) {
			$msg = $envelope-&gt;getBody();
			$this-&gt;logging-&gt;debug('Protocol: '.$msg.' ');
			//$result = $this-&gt;loader($msg);
			$this-&gt;pool-&gt;submit(new RabbitThread($this-&gt;queueName,

new \framework\log\Logging($this->logqueue, $debug=true), $msg));
KaTeX parse error: Expected 'EOF', got '&' at position 7: queue-&̲gt;ack(envelope->getDeliveryTag());
});
$this->channel->qos(0,1);
}
catch(\AMQPConnectionException $e){
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;logging-&gt;…e->__toString());
}
catch(\Exception $e){
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;logging-&gt;…e->__toString());
$connection->disconnect();
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;pool-&gt;shu…tag, $msg){
KaTeX parse error: Expected 'EOF', got '&' at position 6: this-&̲gt;logging-&gt;…msg);
throw new \Exception(tag.′:′.tag.&#x27;: &#x27;.msg);
}

public function __destruct() {
}

}

	</span></strong></pre>
        </div>
            </div>

相關推薦

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

PHP高階程式設計之守護程序，實現優雅重啟

影音先鋒電影http://www.iskdy.com/ 1. 什麼是守護程序 守護程序是脫離於終端並且在後臺執行的程序。守護程序脫離於終端是為了避免程序在執行過程中的資訊在任何終端上顯示並且程序也不會被任何終端所產生的終端資訊所打斷。 例如 apache, nginx, mysql 都是守護

Java程式設計師從笨鳥到菜鳥（五十四） 分散式之訊息佇列

##目錄 該文只是一個複習思路，不瞭解訊息佇列的人建議先看《訊息佇列從入門到精通》 ##一、為什麼使用訊息佇列 三個最主要的應用場景：解耦、非同步、削峰 1、解耦 傳統模式： 缺點：系統間的耦合性太強，如上圖示，系統 A 在程式碼中直接呼叫系統 B 和

訊息中介軟體--RabbitMQ學習（十七）---高階特性之死信佇列

死信佇列:DLX,Dead- Letter- Exchange 利用DLX,當訊息在一個佇列中變成死信( dead message)之後它能被重新 publish到另一個 Exchange,這個 Exchange就是DLX 死信佇列訊息變成死信有一下幾種情況

UNIX網路程式設計：IPC之訊息佇列

訊息佇列：將訊息佇列按佇列的方式組織成的連結串列，每個訊息都是其中的一個節點； 注意：訊息佇列的長度及每個訊息的大小是有限制的 訊息佇列的操作函式如下： msgget 　　int msgget(key_t key,int msgflg); 　

javascript高階程式設計之---繼承

物件的繼承  1.繼承的思想：關鍵是將一個建構函式A的原型指向給另一個建構函式B的例項本身，那麼A例項將擁有（繼承）B例項的所有的屬性和方法（繼承在原型）；   2.原理：將一個建構函式A的原型指向給另一個建構函式B的例項本身，也就意味著A原型將繼承B例項本身擁有的所

javascript高階程式設計之--自執行函式

什麼是自執行函式？ 顧名思義.就是不用呼叫，自己自動就會執行的函式； 為什麼要用自執行函式？ 自執行函式內部是一個獨立的作用域，這樣就保持了一個相對獨立的名稱空間，避免汙染全域性作用域 怎麼建立自執行函式？ 寫法一:(function(param){…})(re—param)

JavaScript高階程式設計之客戶端檢測

瀏覽器檢測，是個很讓人頭痛但又必須要考慮的一個問題。畢竟市面上的瀏覽器廠商眾多，雖然有些標準化的東西約束著他們，但是有些東西，他們還是各自按自己的意願來。這其中以IE瀏覽器最為突出，它幾乎可以被稱之為萬惡之源。 瀏覽器檢測的方法有很多，這裡介紹的可能並不是全部，但基本上都是比較常用的方法。下邊我就來詳細介紹

JavaScript高階程式設計之事件

當用戶按下某個按鈕是，就會觸發對應事件，事件在執行註冊事件時，繫結的程式，實現一次成功的瀏覽器與JS之間的互動。其中事件基於設計模式中的觀察者模式原理實現，但又有些不同，事件它還擁有著一些屬於自己的特點。 事件流 　　事件冒泡是IE的產物，它的流向，是自下向上，從繫結事件的具體元素開始，一直流到具體元素的

Redis之訊息佇列

SpringBoot 結合redis實現訊息佇列功能 釋出者： 配置連線工廠 @Bean public StringRedisTemplate template(RedisConnectionFactory connectionFactory){ retu

PHP高階特性之檔案處理

檔案系統函式用法詳述 1.基本的判斷函式 is_dir — 判斷給定檔名是否是一個目錄 is_file — 判斷給定檔名是否為一個檔案 is_executable — 判斷給定檔名是否可執行 is_link — 判斷給定檔名是

日誌系統之訊息佇列的應用

kafka的概述 Kakfa是由LinkedIn公司開發的一個分散式的訊息系統，後成為Apache頂級開源專案，它使用Scala編寫，以可水平擴充套件和高吞吐率的特性而被廣泛使用。 ps: 通過上述百度百科的概述，只知其然而不知其所以然。 kafka 大家都知道是訊息佇列，那麼籠統的說訊息佇列是

php程序間通訊--訊息佇列

首先我們來看一下如何建立一個訊息佇列。 //建立訊息佇列 $msg_key = ftok( __FILE__, 'a' ); $msg_queue = msg_get_queue( $msg_key, 0666 ); 　在php中通過這兩句話就可以建立一個訊息佇列。 ftok 函式，是可以

Cris 玩轉大資料系列之訊息佇列神器 Kafka

Cris 玩轉大資料系列之訊息佇列神器 Kafka Author：Cris 文章目錄 Cris 玩轉大資料系列之訊息佇列神器 Kafka Author：Cris 1. Kafka 概述

php高階應用之程序控制及程序間通訊

很少有用php寫服務的，然而有些場景又要求能有一個這樣的伺服器程式，它能夠與php無縫結合，並且提供高可靠靠效能的服務，並且提供現有架構所沒有的一些高階特性，例如支援自定義協議，支援長連線等等。PPM(PHP-Process-Manager)是我用PHP開發的一款程序管理

Java併發程式設計 之 同步佇列與等待佇列

在上一篇部落格中，我簡單的介紹了對Condition和ReentrantLock的使用，但是想要更好的掌握多執行緒程式設計，單單會用是不夠的。這篇我會針對Condition方法中的await和signal的實現原理來梳理一下我的理解。 首先我們需要了解同步佇列和等待佇列的概念。簡單的

Java併發程式設計 之 阻塞佇列和CountDownLatch

前幾天看到一個面試題目：有一個長度為2000的字串，開三個執行緒去判斷字串中”u51”的個數。 當時看到這個題目的時候，對併發程式設計是沒有什麼經驗的，在實際專案多執行緒的應用也只有一兩次。最近在惡補《Java併發程式設計的藝術》，對這個題目就有了解題的思路了。在這裡記錄一下對該題的

Python GUI程式設計之訊息傳遞機制&使用者登入

PythonGUI程式設計之-訊息傳遞機制及簡單的使用者登入 訊息機制 1.訊息的傳遞機制 自動發出事件/訊息 訊息由系統負責傳送到佇列 有相關元件進行繫結/設定 後端自動選擇感興趣的事件並作出相應的

php使用redis作為訊息佇列--守護程序模式執行

  出隊指令碼 編寫指令碼，通過crontab定時執行指令碼，從list的右邊取出錯誤處理。 這種方式的優點是不會出現資料的丟失，只有定時指令碼處理完相關資訊之後，才從佇列中取出；缺點也是顯而易見的，響應不夠及時，因為是定時訪問redis的list隊形，所以不能實時的處理佇列中