hiredis XREAD 实例

痛定思痛。 2022-09-16 13:22 87阅读 0赞

#include "stdio.h"
    #include "stdlib.h"
    #include "string.h"
    #include "hiredis.h"
    #include "redisCli.h"
    
    
    #define REDIS_STREAM_KEY       "setMessage" 
    #define REDIS_STREAM_MAX_LEN   1024
    
    static char redisSerIp[64] = {"127.0.0.1"};
    static unsigned short redisSerPort = 6379;
    
    static redisReply *redisStreamRceiveMsg(redisReply *reply)
    {
    	int i = 0;
    	redisReply *childReply = NULL;
    	redisReply *retReply = NULL;
    	
    	while(reply)
    	{
    		if(reply->type == REDIS_REPLY_ARRAY)
    		{
    			childReply = NULL;
    			for (i = 0; i < reply->elements; i++)
    			{
    				childReply = reply->element[i];
    			}
    			
    			if(childReply && childReply->element)
    			{
    				retReply = reply;
    				reply = childReply;
    			}	
    			else
    			{
    				return retReply;
    			}
    		}
    		else
    		{
    			return NULL;
    		}
    	}
    	return NULL;
    }
    
    
    
    
    static void redisStreamDel(redisContext *context,char *cpId)
    {
    	redisReply *reply = NULL;
    	
    	if(context == NULL || cpId == NULL)
    	{
    		return ;
    	}
    	
    	reply = redisCommand(context, "XDEL %s %s",REDIS_STREAM_KEY,cpId);
    	if((reply == NULL) || (context->err))
    	{
    		redisPublishZlogPirnt(ZLOG_ERROR,"[ctl] redisCommand reply is error,XDEL %s %s",
    				REDIS_STREAM_KEY,cpId);
    		if(reply)
    		{
    			freeReplyObject(reply);	
    		}
    		return;
    	}	
    	freeReplyObject(reply);	
    }
    
    
    
    static redisReply *redisStreamRceiveMsgFormat(redisContext *context,redisReply *reply)
    {
    	int i = 0;
    	redisReply *retReply = NULL;
    	redisReply *childReply = NULL;
    	
    	retReply = redisStreamRceiveMsg(reply);
    	if(retReply)
    	{
    		for (i = 0; i < retReply->elements; i++)
    		{
    			childReply = retReply->element[i];
    			if ((childReply->type == REDIS_REPLY_STRING)
    					&& (childReply->str))	
    			{
    				redisPublishZlogPirnt(ZLOG_DEBUG,"[ctl] redisStreamRceiveMsgFormat DEL STR = %s",
    						childReply->str);
    				redisStreamDel(context,childReply->str);
    			}
    		}
    		
    		if(childReply)
    		{
    			return childReply;
    		}
    	}
    	return NULL;
    }
    
    
    
    static redisContext *redisConnInit(void )
    {
    	redisContext *context = NULL;
    	redisReply *reply = NULL;
    	
    	context = redisConnect(redisSerIp, redisSerPort);  //链接本地127.0.0.1，端口为6379的redis
    	if(context != NULL && context->err)
    	{
    		printf("[zlog] redisConnect failed,server : %s, port:%d , SUBSCRIBE channel:zlog\n",redisSerIp,redisSerPort);
    		redisFree(context);
    		return NULL;
    	}	
    	/*控制消息队列的大小*/
    	reply = redisCommand(context, "XTRIM %s MAXLEN %d",REDIS_STREAM_KEY,REDIS_STREAM_MAX_LEN);
    	if((reply == NULL) || (context->err))
    	{
    		redisPublishZlogPirnt(ZLOG_ERROR,"[ctl] redisCommand reply is error,XTRIM %s MAXLEN %d ",
    				REDIS_STREAM_KEY,REDIS_STREAM_MAX_LEN);
    		if(reply)
    		{
    			freeReplyObject(reply);	
    		}				
    		return NULL;
    	}	
    	freeReplyObject(reply);	
    	return context;
    }
    			
    			
    			
    int main(void )
    {
    	int i = 0;
    	redisContext *context = NULL;
    	redisReply *reply = NULL;
    	redisReply *retReply = NULL;
    	redisReply *childReply = NULL;
    	
    	redisPublishZlogPirnt(ZLOG_DEBUG,"[ctl] ctl main start ... ");	
    	context = redisConnInit();
    	if(!context)
    	{
    		return -1;
    	}
    	redisPublishZlogPirnt(ZLOG_DEBUG,"[ctl] redisConnInit ok ... ");	
    
    	while(1)
    	{
    		reply = redisCommand(context, "XREAD block 1000 streams %s $",REDIS_STREAM_KEY);
    		if((reply == NULL) || (context->err))
    		{
    			redisPublishZlogPirnt(ZLOG_ERROR,"[ctl] redisCommand reply is error,XREAD block 1000 streams %s $",
    					REDIS_STREAM_KEY);
    			if(reply)
    			{
    				freeReplyObject(reply);	
    			}					
    			continue;
    		}
    
    		redisPublishZlogPirnt(ZLOG_DEBUG,"[ctl] XREAD block 1000 streams %s $ ok ... ",REDIS_STREAM_KEY);
    		retReply = redisStreamRceiveMsgFormat(context,reply);
    		if(retReply)
    		{
    			for (i = 0; i < retReply->elements; i++)
    			{ 
    				childReply = retReply->element[i];
    				if ((childReply->type == REDIS_REPLY_STRING)
    						&& (childReply->str))	
    				{
    					printf("str = %s\r\n",childReply->str);
    				}
    			}				
    		}
    		freeReplyObject(reply);			
    	}
    	return 0;
    }