1.基本介绍
按照ZooKeeper典型应用场景一览里的说法,分布式队列有两种,一种是常规的先进先出队列,另一种是要等到队列成员聚齐之后的才统一按序执行。
第二种队列可以先建立一个/queue,赋值为n,表达队列的大小。然后每个队列成员加入时,就判断是否达到队列要求的大小,如果是可以进行下一步动作,否则继续等待队列成员的加入。比较典型的情况是,当一个大的任务可能需要很多的子任务完成才能开始进行。
比如汇总账单的时候,就必须先将用户的消费数据,积分数据等都统计完成后才能开始。汇总账单的程序建立一个队列/Queue,赋值为2,然后分别统计消费数据和积分数据的程序当完成任务时就往/Queue下创建一个临时节点。而汇总账单程序监测到/Queue的子节点个数为2时,就可以开始执行任务了。
实际上,我们也可以先建立一个数目为2的子节点。当一个子任务完成的时候,就删除一个子节点,当所有子节点都被删除的时候,主任务就可以开始执行了。这个过程可以形象的理解为拆除屏障。因此这种队列还有一个专门的词语描述,叫做屏障(barrier)。
2.场景分析
讲了那么多的关于屏障的认识,但是并不打算就去实现它,并且Zookeeper的官方文档也有相关的知识。这次的主要目标是常规的FIFO队列。我将实现队列的两个主要操作:push和pop。
1).
int push(zhandle_t *zkhandle,const char *path,char *element)
-
zkhandle
为zookeeper_init
初始化后的句柄 -
path
为队列的路径 -
element
为要压入队列的内容
2).
int pop(zhandle_t *zkhandle,const char *path,char *element_buffer,int *buffer_len)
-
zkhandle
为zookeeper_init
初始化后的句柄 -
path
为队列的路径 -
element_buffer
为要弹出的缓冲区 -
buffer_len
为指向缓冲区的大小的指针
简单来说,假设队列的路径为/Queue,push
就是就是创建一个临时有序的/Queue/queue-节点。pop就是取出/Queue/下序列号最小的节点。
我们知道在C++中stl里有一个queue的类,实现了push,pop等操作,然而它是非线程安全的,即多个线程同时push/pop的时候可能会出现错误。而由于ZooKeeper保证了创建节点和删除节点的一致性,因此可以说利用Zookeeper实现的队列是进程安全的。
3. 场景实践
来看push和pop的具体实现。push
的实现很简单,就是在{path}下创建一个有序的{path}/queue-子节点.
int push(zhandle_t *zkhandle,const char *path,char *element) { char child_path[512] = {0}; char path_buffer[512] = {0}; int bufferlen = sizeof(path_buffer); sprintf(child_path,"%s/queue-",path); int ret = zoo_create(zkhandle,child_path,element,strlen(element), &ZOO_OPEN_ACL_UNSAFE,ZOO_SEQUENCE, path_buffer,bufferlen); if(ret != ZOK){ fprintf(stderr,"failed to create the path %s!\n",path); }else{ printf("create path %s successfully!\n",path); } return ret; }
pop
的功能则是取出{path}下序号最小的子节点,如果没有子节点,则返回-1.
int pop(zhandle_t *zkhandle,const char *path,char *element,int *len) { int i = 0; struct String_vector children; int ret = zoo_get_children(zkhandle,path,0,&children); if(ret != ZOK){ fprintf(stderr,"failed to create the path %s!\n",path); }else if (children.count == 0){ strcpy(element,""); *len = 0; ret = -1; }else{ char *min = children.data[0]; for(i = 0; i < children.count; ++i){ printf("%s:%s\n",min,children.data[i]); if(strcmp(min,children.data[i]) > 0){ min = children.data[i]; } } if(min != NULL){ char child_path[512]={0}; sprintf(child_path,"%s/%s",path,min); ret = zoo_get(zkhandle,child_path,0,element,len,NULL); if(ret != ZOK){ fprintf(stderr,"failed to get data of the path %s!\n",child_path); }else{ ret = zoo_delete(zkhandle,child_path, -1); if(ret != ZOK){ fprintf(stderr,"failed to delete the path %s!\n",child_path); } } } } for(i = 0; i < children.count; ++i){ free(children.data[i]); children.data[i] = NULL; } return ret; }
最后,再来看看模拟队列操作的程序。和其他程序类似,它的选项有
-
-p
:指定队列的路径 -
-m
:指定操作是push还是pop -
-v
:只在push时有用,用与指定要push的元素的值 -
-s
:指定Zookeeper的服务器的ip:port.
如:
向队列/Queue中压人一个元素,元素的值为"Hello":
>myqueue -s 172.17.0.36:2181 -p /Queue -m push -v Hello
将队列/Queue弹出一个元素
>myqueue -s 172.17.0.36:2181 -p /Queue -m pop
最后附上完整的源代码:
#include<stdio.h> #include<string.h> #include<unistd.h> #include"zookeeper.h" #include"zookeeper_log.h" char g_host[512]= "172.17.0.36:2181"; char g_path[512]= "/Queue"; char g_value[512]="msg"; enum MODE{PUSH_MODE,POP_MODE} g_mode; void print_usage(); void get_option(int argc,const char* argv[]); /**********unitl*********************/ void print_usage() { printf("Usage : [myqueue] [-h] [-m mode] [-p path ] [-v value][-s ip:port] \n"); printf(" -h Show help\n"); printf(" -p Queue path\n"); printf(" -m mode:push or pop\n"); printf(" -v the value you want to push\n"); printf(" -s zookeeper server ip:port\n"); printf("For example:\n"); printf(" push the message \"Hello\" into the queue Queue:\n"); printf(" >myqueue -s172.17.0.36:2181 -p /Queue -m push -v Hello\n"); printf(" pop one message from the queue Queue:\n"); printf(" >myqueue -s172.17.0.36:2181 -p /Queue -m pop\n"); } void get_option(int argc,const char* argv[]) { extern char *optarg; int optch; int dem = 1; const char optstring[] = "hv:m:p:s:"; g_mode = PUSH_MODE; while((optch = getopt(argc , (char * const *)argv , optstring)) != -1 ) { switch( optch ) { case 'h': print_usage(); exit(-1); case '?': print_usage(); printf("unknown parameter: %c\n", optopt); exit(-1); case ':': print_usage(); printf("need parameter: %c\n", optopt); exit(-1); case 'm': if(strcasecmp(optarg,"push")==0){ g_mode = PUSH_MODE; }else{ g_mode = POP_MODE; } break; case 's': strncpy(g_host,optarg,sizeof(g_host)); break; case 'p': strncpy(g_path,optarg,sizeof(g_path)); break; case 'v': strncpy(g_value,optarg,sizeof(g_value)); break; default: break; } } } int push(zhandle_t *zkhandle,const char *path,char *element) { char child_path[512] = {0}; char path_buffer[512] = {0}; int bufferlen = sizeof(path_buffer); sprintf(child_path,"%s/queue-",path); int ret = zoo_create(zkhandle,child_path,element,strlen(element), &ZOO_OPEN_ACL_UNSAFE,ZOO_SEQUENCE, path_buffer,bufferlen); if(ret != ZOK){ fprintf(stderr,"failed to create the path %s!\n",path); }else{ printf("create path %s successfully!\n",path); } return ret; } int pop(zhandle_t *zkhandle,const char *path,char *element,int *len) { int i = 0; struct String_vector children; int ret = zoo_get_children(zkhandle,path,0,&children); if(ret != ZOK){ fprintf(stderr,"failed to create the path %s!\n",path); }else if (children.count == 0){ strcpy(element,""); *len = 0; ret = -1; }else{ char *min = children.data[0]; for(i = 0; i < children.count; ++i){ printf("%s:%s\n",min,children.data[i]); if(strcmp(min,children.data[i]) > 0){ min = children.data[i]; } } if(min != NULL){ char child_path[512]={0}; sprintf(child_path,"%s/%s",path,min); ret = zoo_get(zkhandle,child_path,0,element,len,NULL); if(ret != ZOK){ fprintf(stderr,"failed to get data of the path %s!\n",child_path); }else{ ret = zoo_delete(zkhandle,child_path, -1); if(ret != ZOK){ fprintf(stderr,"failed to delete the path %s!\n",child_path); } } } } for(i = 0; i < children.count; ++i){ free(children.data[i]); children.data[i] = NULL; } return ret; } int front(zhandle_t *zkhandle,char *path,char *element,int *len) { int i = 0; struct String_vector children; int ret = zoo_get_children(zkhandle,path,0,&children); if(ret != ZOK){ fprintf(stderr,"failed to create the path %s!\n",path); }else if(children.count == 0){ strcpy(element,""); *len = 0; ret = -1; }else{ char *min = NULL; for(i = 0; i < children.count; ++i){ if(strcmp(min,children.data[i]) > 0){ min = children.data[i]; } } if(min != NULL){ char child_path[512]={0}; sprintf(child_path,"%s/%s",path,min); ret = zoo_get(zkhandle,child_path,0,element,len,NULL); if(ret != ZOK){ fprintf(stderr,"failed to get data of the path %s!\n",child_path); } } } for(i = 0; i < children.count; ++i){ free(children.data[i]); children.data[i] = NULL; } return ret; } int main(int argc, const char *argv[]) { int timeout = 30000; char path_buffer[512]; int bufferlen=sizeof(path_buffer); zoo_set_debug_level(ZOO_LOG_LEVEL_WARN); //设置日志级别,避免出现一些其他信息 get_option(argc,argv); zhandle_t* zkhandle = zookeeper_init(g_host,NULL, timeout, 0, (char *)"lock Test", 0); if (zkhandle ==NULL) { fprintf(stderr, "Error when connecting to zookeeper servers...\n"); exit(EXIT_FAILURE); } int ret = zoo_exists(zkhandle,g_path,0,NULL); if(ret != ZOK){ ret = zoo_create(zkhandle,g_path,"1.0",strlen("1.0"), &ZOO_OPEN_ACL_UNSAFE,0, path_buffer,bufferlen); if(ret != ZOK){ fprintf(stderr,"failed to create the path %s!\n",g_path); }else{ printf("create path %s successfully!\n",g_path); } } if(g_mode == PUSH_MODE){ push(zkhandle,g_path,g_value); printf("push:%s\n",g_value); }else{ int len = sizeof(g_value); ret = pop(zkhandle,g_path,g_value,&len) ; if(ret == ZOK){ printf("pop:%s\n",g_value); }else if( ret == -1){ printf("queue is empty\n"); } } zookeeper_close(zkhandle); return 0; }