流streams · PHP筆記

~~~ 5.0 全新數據類型數據流streams 看起來比pubsub可靠多的消息隊列。pubsub不靠譜？很不靠譜，網絡一斷或buffer一大就會主動清理數據。stream的設計參考了kafka的消費組模型以更抽象的方式建模日志的數據結構。Redis的streams主要是一個append only的數據結構，至少在概念上它是一種在內存中表示的抽象數據類型，只不過它們實現了更強大的操作，以克服日志文件本身的限制如果你了解MQ，那么可以把streams當做MQ。如果你還了解kafka，那么甚至可以把streams當做kafka 這個功能有點類似于redis以前的Pub/Sub，但是也有基本的不同： streams支持多個客戶端（消費者）等待數據（Linux環境開多個窗口執行XREAD即可模擬），并且每個客戶端得到的是完全相同的數據 Pub/Sub是發送忘記的方式，并且不存儲任何數據；而streams模式下，所有消息被無限期追加在streams中，除非用于顯示執行刪除（XDEL） streams的Consumer Groups也是Pub/Sub無法實現的控制方式 streams數據結構本身非常簡單，但是streams依然是Redis到目前為止最復雜的類型，其原因是實現的一些額外的功能：一系列的阻塞操作允許消費者等待生產者加入到streams的新數據。另外還有一個稱為Consumer Groups的概念，這個概念最先由kafka提出，Redis有一個類似實現，和kafka的Consumer Groups的目的是一樣的：允許一組客戶端協調消費相同的信息流 stream適用于允許丟失數據的業務場景，因為redis本身是不支持數據的絕對可靠的，哪怕aof調成always Redis Stream——作為消息隊列的典型應用場景 https://segmentfault.com/a/1190000016777728 xAck($stream, $group, $arr_messages_ids) - 確認一條或多條待處理的消息 $redis->xAck('mystream_key', 'group1', ['1530063064286-0', '1530063064286-1']); xAdd($str_key, $str_id(* 代表redis自動生成流序列id), $arr_message) - 向流添加消息返回添加消息的id $obj_redis->xAdd('mystream_key', "*", ['field' => 'value']);//1530063064286-0 xClaim($str_key, $str_group, $str_consumer, $min_idle_time, $arr_messages_ids, [$arr_options]) - 獲取待處理信息的所有權返回消息ID數組以及相應的數據 $ids = ['1530113681011-0', '1530113681011-1', '1530113681011-2']; // 無options $obj_redis->xClaim( 'mystream', 'group1', 'myconsumer1', 0, $ids ); // 有 options $obj_redis->xClaim( 'mystream', 'group1', 'myconsumer2', 0, $ids, [ 'IDLE' => time() * 1000, 'RETRYCOUNT' => 5, 'FORCE', 'JUSTID' ] ); xDel(stream_key, $arr_messages_ids) - 從流中刪除消息 $redis->xDel('mystream', ['1530115304877-0', '1530115305731-0']); Group - 創建,銷毀或管理消費者群組 //創建 $redis->xGroup('CREATE', $str_key, $str_group, $str_msg_id, [$boo_mkstream]); //刪除 $redis->xGroup('DESTROY', $str_key, $str_group); //管理 $obj_redis->xGroup('HELP'); $obj_redis->xGroup('SETID', $str_key, $str_group, $str_msg_id); $obj_redis->xGroup('DELCONSUMER', $str_key, $str_group, $str_consumer_name); //刪除消費者 xInfo('CONSUMERS/GROUPS/STREAM/HELP', $str_stream, $str_group) - 獲取有關流的信息 xLen(stream_key) - 獲取流的總數據個數 xPending($str_stream, $str_group [, $str_start, $str_end, $i_count, $str_consumer]) - 檢查流中的待處理消息 $redis->xPending('mystream', 'mygroup', '-', '+', 1, 'consumer-1'); xRange($str_stream, $str_start, $str_end [, $i_count]) - 查詢流中的一系列消息 $redis->xRange('mystream', '-', '+', 2); xRead($arr_streams [, $i_count, $i_block) - 從流中讀取消息 $redis->xRead(['stream1' => '1535222584555-0', 'stream2' => '1535222584555-0']); //僅接收新消息(($=last id)并無限期等待一個新消息 $redis->xRead(['stream1' => '$'], 1, 0); // xReadGroup($str_group, $str_consumer, $arr_streams [, $i_count, $i_block]) - 使用組和消費者一起讀取流消息類似于xRead,但它支持讀取特定消費者組的消息返回傳遞給此使用者組的消息（如果有） $redis->xReadGroup('mygroup', 'consumer2', ['s1' => 0, 's2' => 0], 1, 1000); xRevRange($str_stream, $str_end, $str_start [, $i_count]) - 查詢從start到end的一條或多條消息 $redis->xRevRange('mystream', '+', '-'); xTrim($str_stream, $i_max_len [, $boo_approximate])- 流裁剪為指定數量的項目 $obj_redis->xTrim('mystream', 100, true); ~~~