Redis 中使用 list,streams,pub/sub 几种方式实现消息队列的问题_Redis_

127.0.0.1:6379> LPUSH test "ceshi-1" (integer) 1 127.0.0.1:6379> RPOP test "ceshi-1" 

127.0.0.1:6379> BRPOP test 10 

127.0.0.1:6379> LPUSH test "ceshi-1" (integer) 1 127.0.0.1:6379> LPUSH test "ceshi-2" (integer) 2 127.0.0.1:6379> BRPOPLPUSH test a-test 100 "ceshi-1" 127.0.0.1:6379> BRPOPLPUSH test a-test 100 "ceshi-2" 127.0.0.1:6379> BRPOPLPUSH test a-test 100 127.0.0.1:6379> RPOP a-test "ceshi-1" 127.0.0.1:6379> RPOP a-test "ceshi-2" 

void listTypePush(robj *subject, robj *value, int where) { if (subject->encoding == OBJ_ENCODING_QUICKLIST) { int pos = (where == LIST_HEAD) ? QUICKLIST_HEAD : QUICKLIST_TAIL; if (value->encoding == OBJ_ENCODING_INT) { char buf[32]; ll2string(buf, 32, (long)value->ptr); quicklistPush(subject->ptr, buf, strlen(buf), pos); } else { quicklistPush(subject->ptr, value->ptr, sdslen(value->ptr), pos); } } else { serverPanic("Unknown list encoding"); } } /* Wrapper to allow argument-based switching between HEAD/TAIL pop */ void quicklistPush(quicklist *quicklist, void *value, const size_t sz, int where) { if (where == QUICKLIST_HEAD) { quicklistPushHead(quicklist, value, sz); } else if (where == QUICKLIST_TAIL) { quicklistPushTail(quicklist, value, sz); } } 

typedef struct quicklist { // quicklist的链表头 quicklistNode *head; // quicklist的链表尾 quicklistNode *tail; // 所有ziplist中的总元素个数 unsigned long count; /* total count of all entries in all ziplists */ // quicklistNodes的个数 unsigned long len; /* number of quicklistNodes */ int fill : QL_FILL_BITS; /* fill factor for individual nodes */ unsigned int compress : QL_COMP_BITS; /* depth of end nodes not to compress;0=off */ unsigned int bookmark_count: QL_BM_BITS; quicklistBookmark bookmarks[]; } quicklist; typedef struct quicklistNode { // 前一个quicklistNode struct quicklistNode *prev; // 后一个quicklistNode struct quicklistNode *next; // quicklistNode指向的ziplist unsigned char *zl; // ziplist的字节大小 unsigned int sz; /* ziplist size in bytes */ // ziplist中的元素个数 unsigned int count : 16; /* count of items in ziplist */ // 编码格式，原生字节数组或压缩存储 unsigned int encoding : 2; /* RAW==1 or LZF==2 */ // 存储方式 unsigned int container : 2; /* NONE==1 or ZIPLIST==2 */ // 数据是否被压缩 unsigned int recompress : 1; /* was this node previous compressed? */ // 数据能否被压缩 unsigned int attempted_compress : 1; /* node can't compress; too small */ // 预留的bit位 unsigned int extra : 10; /* more bits to steal for future usage */ } quicklistNode; 

/* Add new entry to head node of quicklist. * * Returns 0 if used existing head. * Returns 1 if new head created. */ int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) { quicklistNode *orig_head = quicklist->head; assert(sz < UINT32_MAX); /* TODO: add support for quicklist nodes that are sds encoded (not zipped) */ if (likely( // 检测插入位置的 ziplist 是否能容纳该元素 _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) { quicklist->head->zl = ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD); quicklistNodeUpdateSz(quicklist->head); } else { // 容纳不了，就重新创建一个 quicklistNode quicklistNode *node = quicklistCreateNode(); node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD); quicklistNodeUpdateSz(node); _quicklistInsertNodeBefore(quicklist, quicklist->head, node); } quicklist->count++; quicklist->head->count++; return (orig_head != quicklist->head); } 

XADD：插入消息，保证有序，可以自动生成全局唯一ID； XREAD：用于读取消息，可以按ID读取数据； XREADGROUP：按消费组形式读取消息； XPENDING和XACK：XPENDING命令可以用来查询每个消费组内所有消费者已读取但尚未确认的消息，而XACK命令用于向消息队列确认消息处理已完成。 

$ XADD key ID field value [field value ...] 

$ XADD teststream * name xiaohong surname xiaobai "1646650328883-0" 

$ XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...] 

$ XREAD BLOCK 100 STREAMS teststream 0 1) 1) "teststream" 2) 1) 1) "1646650328883-0" 2) 1) "name" 2) "xiaohong" 3) "surname" 4) "xiaobai" 

$ XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername] 

$ XGROUP CREATE teststream test-consumer-group-name 0-0 

$ XGROUP CREATE teststream test-consumer-group-name $ 

$ XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...] 

$ XADD teststream * name xiaohong surname xiaobai "1646653392799-0" $ XREADGROUP GROUP test-consumer-group-name test-consumer-name COUNT 1 STREAMS teststream > 1) 1) "teststream" 2) 1) 1) "1646653392799-0" 2) 1) "name" 2) "xiaohong" 3) "surname" 4) "xiaobai" 

$ XPENDING teststream test-consumer-group-name 1) (integer) 3 2) "1646653325535-0" 3) "1646653392799-0" 4) 1) 1) "test-consumer-name" 2) "3" 

typedef struct stream { // 这是使用前缀树存储数据 rax *rax; /* The radix tree holding the stream. */ uint64_t length; /* Number of elements inside this stream. */ // 当前stream的最后一个id streamID last_id; /* Zero if there are yet no items. */ // 存储当前的消费者组信息 rax *cgroups; /* Consumer groups dictionary: name -> streamCG */ } stream; t
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：Redis中有序集合的内部实现方式的详细介绍_Redis_
                                                                
下一篇：Redis 的查询很快的原因解析及Redis 如何保证查询的高效_Redis_