コミュニティ 学ぶ ツールライブラリ レジャー
検索
日本語
ホームページ > バックエンド開発 > PHPチュートリアル > 本文

PHPメッセージキュー

WBOY
リリース： 2016-06-13 13:11:32
オリジナル
908 人が閲覧しました

PHPメッセージキュー

php-メッセージキューを実装し、共有メモリを介して通信を処理する 2 つのクラス

メッセージ キューを実装するには、Apache ActiveMQ、memcacheq... などのより専門的なツールを使用できます。ここでは 2 つの基本的で簡単な実装方法を示します。

memcache メソッドを使用して実現します 

<?php
/*
 * @Copyright (c) 2007,上海友邻信息科技有限公司
 * @All	rights reserved.
 *
 * 这个消息队列不是线程安全的，我只是尽量的避免了冲突的可能性。如果你要实现线程安全的，一个建议是通过文件进行锁定，然后进行操作。
 *
 * @filename   MemcacheQueue.class.php
 */
/**
 * Class and Function List:
 * Function list:
 * - __construct()
 * - singleton()
 * - init()
 * - __get()
 * - __set()
 * - isEmpty()
 * - isFull()
 * - enQueue()
 * - deQueue()
 * - getTop()
 * - getAll()
 * - getPage()
 * - makeEmpty()
 * - getAllKeys()
 * - add()
 * - increment()
 * - decrement()
 * - set()
 * - get()
 * - delete()
 * - getKeyByPos()
 * Classes list:
 * - Yl_MemcacheQueue
 */

class Yl_MemcacheQueue {
	private static $instance;
	private $memcache;
	private $name;
	private $prefix;
	private $maxSize;
	
	private function __construct() {
	}
	
	static function singleton() {
		
		if (! (self::$instance instanceof self)) {
			self::$instance = new Yl_MemcacheQueue ();
		
		}
		
		return self::$instance;
	}
	
	public function init($max_size, $name, $prefix = "__queue__") {
		$max_size = 1000;
		$name = '_war_';
		$prefix = '_queue';
		
		$this->memcache = Yl_Memcache::singleton ();
		$this->name = $name;
		$this->prefix = $prefix;
		$this->maxSize = $max_size;
		
		$this->add ( 'front', 0 );
		$this->add ( 'rear', 0 );
		$this->add ( 'size', 0 );
	}
	
	function isEmpty() {
		return $this->get ( 'size' ) == 0;
	}
	
	function isFull() {
		return $this->get ( 'size' ) >= $this->maxSize;
	}
	
	function enQueue($data) {
		if ($this->isFull ()) {
			throw new Exception ( "Queue is Full" );
		}
		
		$size = $this->increment ( 'size' );
		$rear = $this->increment ( 'rear' );
		
		$this->set ( ($rear - 1) % $this->maxSize, $data );
		
		return $this;
	}
	
	function deQueue() {
		if ($this->isEmpty ()) {
			throw new Exception ( "Queue is Empty" );
		}
		
		$this->decrement ( 'size' );
		$front = $this->increment ( 'front' );
		$this->delete ( ($front - 1) % $this->maxSize );
		
		return $this;
	}
	
	function getTop() {
		return $this->get ( $this->get ( 'front' ) % $this->maxSize );
	}
	
	function getAll() {
		return $this->getPage ();
	}
	
	function getPage($offset = 0, $limit = 0) {
		$size = $this->get ( 'size' );
		
		if (0 == $size || $size < $offset) {
			return null;
		}
		
		$front = $this->get ( 'front' ) % $this->maxSize;
		$rear = $this->get ( 'rear' ) % $this->maxSize;
		
		$keys [] = $this->getKeyByPos ( ($front + $offset) % $this->maxSize );
		$num = 1;
		
		for($pos = ($front + $offset + 1) % $this->maxSize; $pos != $rear; $pos = ($pos + 1) % $this->maxSize) {
			$keys [] = $this->getKeyByPos ( $pos );
			$num ++;
			
			if ($limit > 0 && $limit == $num) {
				break;
			}
		}
		
		return array_values ( $this->memcache->get ( $keys ) );
	}
	
	function makeEmpty() {
		$keys = $this->getAllKeys ();
		
		foreach ( $keys as $value ) {
			$this->delete ( $value );
		}
		
		$this->delete ( "rear" );
		$this->delete ( "front" );
		$this->delete ( 'size' );
		$this->delete ( "maxSize" );
	}
	
	private function getAllKeys() {
		if ($this->isEmpty ()) {
			return array ();
		}
		
		$keys [] = $this->get ( 'front' );
		
		for($pos = ($this->get ( 'front' ) % $this->maxSize + 1) % $this->maxSize; $pos != $this->get ( 'rear' ) % $this->maxSize; $pos = ($pos + 1) % $this->maxSize) {
			$keys [] = $pos;
		}
		
		return $keys;
	}
	
	private function add($pos, $data) {
		$this->memcache->add ( $this->getKeyByPos ( $pos ), $data );
		
		return $this;
	}
	
	private function increment($pos) {
		
		return $this->memcache->increment ( $this->getKeyByPos ( $pos ) );
	}
	
	private function decrement($pos) {
		$this->memcache->decrement ( $this->getKeyByPos ( $pos ) );
	}
	
	private function set($pos, $data) {
		$this->memcache->save ( $data, $this->getKeyByPos ( $pos ) );
		
		return $this;
	}
	
	private function get($pos) {
		
		return $this->memcache->get ( $this->getKeyByPos ( $pos ) );
	}
	
	private function delete($pos) {
		
		return $this->memcache->delete ( $this->getKeyByPos ( $pos ) );
	}
	
	private function getKeyByPos($pos) {
		
		return $this->prefix . $this->name . $pos;
	}
}
?>
ログイン後にコピー
共有メモリキューを使用した実装 クリックして元のアドレスを表示

?

PHP を使用して Linux メッセージ キューを操作し、プロセス間通信を完了します

私たちが開発するシステムをマルチプロセス モードで実行する必要がある場合、プロセス間通信が重要なリンクになります。メッセージキューは、Linux システムにおけるプロセス間通信の方法です。
Linux システムプロセス通信の概念と実装については、http://www.ibm.com/developerworks/cn/linux/l-ipc/
を参照してください。 Linux システムにおけるメッセージキューの概念と実装については、http://www.ibm.com/developerworks/cn/linux/l-ipc/part4/
を参照してください。 PHP の sysvmsg モジュールは、Linux システムでサポートされる System V IPC の System V メッセージ キュー関数ファミリーをカプセル化したものです。プロセス間通信には sysvmsg モジュールが提供する関数を使用する必要があります。まずサンプルコードを見てみましょう_1: 

<?php
$message_queue_key = ftok(__FILE__, 'a');

$message_queue = msg_get_queue($message_queue_key, 0666);
var_dump($message_queue);

$message_queue_status = msg_stat_queue($message_queue);
print_r($message_queue_status);

//向消息队列中写
msg_send($message_queue, 1, "Hello,World!");

$message_queue_status = msg_stat_queue($message_queue);
print_r($message_queue_status);

//从消息队列中读
msg_receive($message_queue, 0, $message_type, 1024, $message, true, MSG_IPC_NOWAIT);
print_r($message."\r\n");

msg_remove_queue($message_queue);
?>
ログイン後にコピー
?このコードを実行した結果は次のとおりです。 
resource(4) of type (sysvmsg queue)
Array
(
    [msg_perm.uid] => 1000
    [msg_perm.gid] => 1000
    [msg_perm.mode] => 438
    [msg_stime] => 0
    [msg_rtime] => 0
    [msg_ctime] => 1279849495
    [msg_qnum] => 0
    [msg_qbytes] => 16384
    [msg_lspid] => 0
    [msg_lrpid] => 0
)
Array
(
    [msg_perm.uid] => 1000
    [msg_perm.gid] => 1000
    [msg_perm.mode] => 438
    [msg_stime] => 1279849495
    [msg_rtime] => 0
    [msg_ctime] => 1279849495
    [msg_qnum] => 1
    [msg_qbytes] => 16384
    [msg_lspid] => 2184
    [msg_lrpid] => 0
)
Hello,World!
ログイン後にコピー
?「Hello, World!」という文字列がメッセージ キューから正常に読み取られたことがわかります。

サンプルコードの主な関数は以下のとおりです: 

ftok ( string $pathname , string $proj ) 
	手册上给出的解释是：Convert a pathname and a project identifier to a System V IPC key。这个函数返回的键值唯一对应linux系统中一个消息队列。在获得消息队列的引用之前都需要调用这个函数。

msg_get_queue ( int $key [, int $perms ] )
	msg_get_queue()会根据传入的键值返回一个消息队列的引用。如果linux系统中没有消息队列与键值对应，msg_get_queue()将会创建一个新的消息队列。函数的第二个参数需要传入一个int值，作为新创建的消息队列的权限值，默认为0666。这个权限值与linux命令chmod中使用的数值是同一个意思，因为在linux系统中一切皆是文件。

msg_send ( resource $queue , int $msgtype , mixed $message [, bool $serialize [, bool $blocking [, int &$errorcode ]]] )
	顾名思义，该函数用来向消息队列中写数据。

msg_stat_queue ( resource $queue ) 
	这个函数会返回消息队列的元数据。消息队列元数据中的信息很完整，包括了消息队列中待读取的消息数、最后读写队列的进程ID等。示例代码在第8行调用该函数返回的数组中队列中待读取的消息数msg_qnum值为0。

msg_receive ( resource $queue , int $desiredmsgtype , int &$msgtype , int $maxsize , mixed &$message [, bool $unserialize [, int $flags [, int &$errorcode ]]] ) 
	msg_receive用于读取消息队列中的数据。

msg_remove_queue ( resource $queue ) 
    msg_remove_queue用于销毁一个队列。
ログイン後にコピー
?サンプルコード_1は、PHPのメッセージキュー機能の応用例を示したものです。次のコードは、プロセス間通信シナリオを具体的に説明しています。 
<?php
$message_queue_key = ftok ( __FILE__, 'a' );
$message_queue = msg_get_queue ( $message_queue_key, 0666 );

$pids = array ();
for($i = 0; $i < 5; $i ++) {
	//创建子进程
	$pids [$i] = pcntl_fork ();
	
	if ($pids [$i]) {
		echo "No.$i child process was created, the pid is $pids[$i]\r\n";
	} elseif ($pids [$i] == 0) {
		$pid = posix_getpid ();
		echo "process.$pid is writing now\r\n";
		
		msg_send ( $message_queue, 1, "this is process.$pid's data\r\n" );
		posix_kill ( $pid, SIGTERM );
	}
}

do {
	msg_receive ( $message_queue, 0, $message_type, 1024, $message, true, MSG_IPC_NOWAIT );
	echo $message;

	//需要判断队列是否为空，如果为空就退出
//break;
} while ( true )
?>
ログイン後にコピー
実行結果は次のとおりです。 
No.0 child process was created, the pid is 5249
No.1 child process was created, the pid is 5250
No.2 child process was created, the pid is 5251
No.3 child process was created, the pid is 5252
No.4 child process was created, the pid is 5253
process.5251 is writing now
this is process.5251's data
process.5253 is writing now
process.5252 is writing now
process.5250 is writing now
this is process.5253's data
this is process.5252's data
this is process.5250's data
process.5249 is writing now
this is process.5249's data
ログイン後にコピー

レデ​​ィス
http://www.neatstudio.com/show-976-1.shtml

?

PHP に付属する 3 つのメッセージ キュー関連関数
http://www.zhangguangda.com/?p=89?

関連ラベル：
gt message msg queue this
ソース：php.cn
前の記事：Firefox では Ajax 互換性に問題がありますが、プログラムは IE では正常に実行されます 次の記事：PHP5 VC9、VC6、スレッド セーフ、非スレッド セーフのバージョン間の違い
このウェブサイトの声明
この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。
著者別の最新記事
最新の問題
Vueでルートプロパティを設定する方法はありますか? ページ内の特定のタブ this.$router.push({name:"AdminNotifications",params:{tab:"reques...
から 2024-04-06 11:35:59
0
1
456
Route::controller(<class>)->group(...) で "resource"; メソッドを使用する アプリケーションのルーティング ファイルを編集して、できるだけ整理しています。そこで、「web.php」でこれを実行しました: Route::controller(Alquiler...
から 2024-04-04 09:09:42
0
1
408
Vuex パラメータの受け渡し this.$store.commit('deleteCheckboxItems',response.data.items.forEach(element=>{element....
から 2024-04-03 22:36:08
0
2
334
配列内に要素が存在するかどうかを判断するプロセスが無効です ご迷惑をおかけして申し訳ありませんが、これを理解するのに2時間ほど費やしました。問題は、コードが 1 つの要素のみを追加し、他の ID を追加しないことです。「すでにある」 $(f...
から 2024-04-03 13:03:54
0
1
377
React コンポーネント インスタンスへのポインターをコンストラクターの外部で取得できるかどうかを調べる 次の React コンポーネントを考えてみましょう: importReact,{Component}from'react'classFrameextendsComponent{co...
から 2024-04-02 22:07:22
0
1
318
関連トピック
詳細>
人気のおすすめ
人気のチュートリアル
詳細>
関連するチュートリアル
人気のおすすめ
最新のコース
最新のダウンロード
詳細>
ウェブエフェクト
公式サイト
サイト素材
フロントエンドテンプレート