Knative 中的消息队列集成和事件处理实践

发布时间: 2023-12-28 10:07:31 阅读量: 26 订阅数: 35

嵌入式实验（消息队列）

#include <string.h> #include <stdio.h> #include "vxWorks.h" #include "msgQLib.h" #include "taskLib.h" /*#include "memPartLib.h"*/ #include "memLib.h" /*宏定义*/ #define MAX_MSGS (10) /* the length of msg*/ #define MAX_MSG_LEN sizeof(MESSAGE) /*the length of message*/ #define STACK_SIZE 20000 /*the stack size of task*/ #define DELAY_TICKS 50 /*the time of sending message*/ #define MAX_point 5 /*用户从系统内存池中获得内存的最大次数*/ #define size_1 30 /*用户分区的分配的大小*/ #define size_2 40 /*全局变量*/ int tidtask1; int tidtask2; int tidtask3; SEM_ID syncSemId; SEM_ID waitSemId; MSG_Q_ID myMsgQId1; MSG_Q_ID myMsgQId2; MSG_Q_ID myMsgQId3; typedef struct _MESSAGE { int mSendId; /*发送任务 ID*/ int mRecvId; /*接收任务 ID*/ int mData; /*消息中传递的数据*/ char Data[14]; } MESSAGE; /*内存管理*/ char* usermem1; char* usermem2; MESSAGE *point1[MAX_point]; MESSAGE *point2[MAX_point]; MESSAGE *point3[MAX_point]; int point1_index=0; int point2_index=0; int point3_index=0; PART_ID partid1; PART_ID partid2; #define MID_MESSAGE(id) (id) /*函数声明*/ int start(void); int task1(void); int task2(void); int task3(void); template<T> T* mymalloc(unsigned nBytes); void myfree(void); void bye(void); /***************************************[progStart]*******************************************/ /*启动程序，创建息队例，任务*/ int start(void) { tidtask1=taskSpawn("tTask1", 220, 0, STACK_SIZE, (FUNCPTR)task1,0,0,0,0,0,0,0,0,0,0); usermem1=malloc(200); partid1=memPartCreate(usermem1,200); usermem2=malloc(400); partid2=memPartCreate(usermem2,400); return; } /**************************************[test_end]********************************************/ /*是否相等，相等返回1*/ int test_end(char *end,char *target) { int ret; if(!strcmp(end,target)) ret=1; else ret=0; return ret; } /****************************************[task1]***********************************************/ /*管理Task。负责系统启动时同步系统中其他Task的启动同步，利用信号量的semFlush()完成。同时接收各*/ /*Task的告警信息，告警信息需编号以logmsg方式输出。本task负责系统结束时的Task删除处理*/ int task1(void) { int singal; int message; MESSAGE *rxMsg=mymalloc<MESSAGE>(26); /*define messages,and alloc memory*/ memset(rxMsg,0,26); syncSemId=semBCreate(SEM_Q_FIFO,SEM_EMPTY); /*creat semaphore*/ waitSemId=semBCreate(SEM_Q_PRIORITY,SEM_EMPTY); myMsgQId1=msgQCreate(MAX_MSGS,MAX_MSG_LEN,MSG_Q_PRIORITY); /*create msgQ*/ myMsgQId2=msgQCreate(MAX_MSGS,MAX_MSG_LEN,MSG_Q_PRIORITY); myMsgQId3=msgQCreate(MAX_MSGS,MAX_MSG_LEN,MSG_Q_PRIORITY); tidtask2=taskSpawn("tTask2", 200, 0, STACK_SIZE, (FUNCPTR)task2,0,0,0,0,0,0,0,0,0,0); /*create task*/ tidtask3=taskSpawn("tTask3", 210, 0, STACK_SIZE, (FUNCPTR)task3,0,0,0,0,0,0,0,0,0,0); printf("Please input one of the following commands:add,sub,multiply,divide,testcommand\n"); /*the command we should put into the console*/ semFlush(syncSemId); /*release semaphore*/ semGive(waitSemId); while(1) { singal=1; msgQReceive(myMsgQId1,(char*)&rxMsg,sizeof(rxMsg),WAIT_FOREVER); if(rxMsg->mRecvId==MID_MESSAGE(3)) /*receive MsgQ from task3*/ { singal=test_end(rxMsg->Data,"wrong length")-1; logMsg("task3 receiveing a:%s\n",rxMsg->Data); /*put the warn from task3*/ logMsg("Please reput the other command!\n"); msgQReceive(myMsgQId1,(char*)&rxMsg,MAX_MSG_LEN,WAIT_FOREVER); /*recive MsgQ from task3*/ } if(rxMsg->mRecvId==MID_MESSAGE(2)) /*receive MsgQ from task2*/ { message=test_end(rxMsg->Data,"sysend"); if(message) { /*if the message from task2 is "sysend" and did not receive the warn from task3, close the system*/ if(singal) { bye(); } } else {/*if the message from task2 is "sysend" and receive the warn from task3, reput the command*/ if(singal) logMsg("task2 receiveing a %s\n",rxMsg->Data); logMsg("please reput the correct command!\n"); } } } return; } /********************************************************************************************/ int change_buf(char *command) { int ret; if(!strcmp(command,"add")) ret=1; else if(!strcmp(command,"sub")) ret=2; else if(!strcmp(command,"multiply")) ret=3; else if(!strcmp(command,"divide")) ret=4; else if(!strcmp(command,"testcommand")) ret=5; else ret=0; return ret; } /****************************************[task2]*********************************************/ /*console 命令行接收Task。接收并分析console发来的命令行及参数。自行设置5种以上命令，并根据命*/ /*令的内容向Task3发送激励消息。同时实现系统退出命令，使系统采用适当方式安全退出。收到非法命令*/ /*向Task1告警*/ int task2(void) { char buf[100]; int command; char *str=mymalloc<char>(35); MESSAGE *txMsg=mymalloc<MESSAGE>(26); memset(str,0,35); memset(txMsg,0,26); txMsg->mSendId=MID_MESSAGE(2); txMsg->mRecvId=MID_MESSAGE(2); FOREVER { semTake(syncSemId,WAIT_FOREVER); semTake(waitSemId,WAIT_FOREVER); gets(buf); command=change_buf(buf);/*change the commands into numbers*/ switch(command) { case 0:/*receive uncorrect command*/ txMsg->mData=0; strcpy(txMsg->Data,"wrong command");/*send warn to task1*/ msgQSend(myMsgQId1,(char*)&txMsg,sizeof(txMsg),WAIT_FOREVER,MSG_PRI_NORMAL); break; case 1:/*receive add command*/ strcpy(str,"This an add caculate!\0"); txMsg->mData=1; break; case 2:/*receive sub command*/ strcpy(str,"This a sub caculate!\0"); txMsg->mData=2; break; case 3:/*receive multiply command*/ strcpy(str,"This a multiply caculate!\0"); txMsg->mData=3; break; case 4:/*receive divide command*/ strcpy(str,"This a divide caculate!\0"); txMsg->mData=4; break; case 5:/*receive testcommand,send a long string to task3*/ strcpy(str,"This a testcommand to warn task1!\0"); txMsg->mData=5; break; default: break; } if(txMsg->mData!=0) {/*send along string to task3,and send a message to taks3*/ msgQSend(myMsgQId3,(char*)&str,sizeof(str),WAIT_FOREVER,MSG_PRI_NORMAL); msgQSend(myMsgQId3,(char*)&txMsg,sizeof(txMsg),WAIT_FOREVER,MSG_PRI_NORMAL); } semGive(waitSemId); semGive(syncSemId); taskDelay(DELAY_TICKS); if(txMsg->mData!=0) {/*send sysend to task1 to let task1 close system*/ strcpy(txMsg->Data,"sysend"); msgQSend(myMsgQId1,(char*)&txMsg,sizeof(txMsg),WAIT_FOREVER,MSG_PRI_NORMAL); } } return; } /****************************************[task3]********************************************/ /*console输出Task。接收需打印输出的字串消息（命令），输出到console。收到长度为0或超常字串向*/ /*Task1告警*/ int task3(void) { int firstData=100; int secondData=10; MESSAGE *rxMsg=mymalloc<MESSAGE>(26); MESSAGE *txMsg=mymalloc<MESSAGE>(26); char *rstr=mymalloc<char>(35); memset(txMsg,0,26); memset(txMsg,0,26); memset(rstr,0,35); txMsg->mSendId=MID_MESSAGE(3); txMsg->mRecvId=MID_MESSAGE(3); while(1) { semTake(syncSemId,WAIT_FOREVER); msgQReceive(myMsgQId3,(char*)&rstr,sizeof(rstr),WAIT_FOREVER); if(strlen(rstr)<=0||strlen(rstr)>=26) {/*make sure whether the string is too long or short*/ strcpy(txMsg->Data,"wrong length"); msgQSend(myMsgQId1,(char*)&txMsg,sizeof(txMsg),WAIT_FOREVER,MSG_PRI_NORMAL); /*msgQReceive(myMsgQId3,(char*)&rxMsg,sizeof(rxMsg),WAIT_FOREVER);*/ } semTake(waitSemId,WAIT_FOREVER); msgQReceive(myMsgQId3,(char*)&rxMsg,sizeof(rxMsg),WAIT_FOREVER); if(rxMsg->mData!=5) {/*when it is not testcommand,printf these*/ printf("%s\n",rstr); printf("there are two datas!\n"); printf("firstData:100\n"); printf("secondData:10\n"); } switch(rxMsg->mData) { case 1:/*printf add caculate*/ printf("The result is:%d\n",firstData+secondData); break; case 2:/*printf sub caculate*/ printf("The result is:%d\n",firstData-secondData); break; case 3:/*printf multiply caculate*/ printf("The result is:%d\n",firstData*secondData); break; case 4:/*printf divide caculate*/ printf("The result is:%d\n",firstData/secondData); break; case 5: break; default: break; } semGive(waitSemId); semGive(syncSemId); taskDelay(DELAY_TICKS); } return; } template<T> T* mymalloc(unsigned nBytes) { T* point; int i=0; /*用户分区一是否能分配的标志位*/ int j=0; /*用户分区二是否能分配的标志位*/ if(nBytes<size_1) { point=memPartAlloc(partid1,size_1); if(point==NULL) { i=1; /*标志位为一，不能再分配*/ printf("the small partid has full or not enough!\n"); printf("we will try to malloc it in the big partid\n"); } else { point1[point1_index]=point; printf("the number of the point1_index is:%d\n",point1_index); point1_index++; } } if (i==1 || (nBytes>=size_1 && nBytes<size_2)) /*若用户分区一不能分配，由分区二来分配*/ { point=memPartAlloc(partid2,size_2); if(point==NULL) { j=1; /*标志位为一，不能再分配*/ printf("the big partid has full or not enough!\n"); printf("we will try to malloc it in system memery\n"); } else { i=0; point2[point2_index]=point; printf("the number of the point2_index is:%d\n",point2_index); point2_index++; } } if((j==1 || nBytes>=size_2) && point3_index<MAX_point) /*若用户分区二不能分配，由系统内存池来分配，且只能从系统内存池中分配MAX_point次*/ { point=malloc(nBytes); point3[point3_index]=point; printf("the number of the point3_index is:%d\n",point3_index); point3_index++; } return point; } void myfree(void) { int i=0; for (i=0;i<point1_index;i++) { memPartFree(partid1,point1[i]); } for (i=0;i<point2_index;i++) { memPartFree(partid2,point2[i]); } for (i=0;i<point3_index;i++) { free(point3[i]); } free(usermem1); free(usermem2); printf("The memory have freed!\n"); } void bye(void) { myfree(); logMsg("Bye-bye\n"); taskDelete(tidtask2); taskDelete(tidtask3); msgQDelete(myMsgQId1); msgQDelete(myMsgQId2); msgQDelete(myMsgQId3); semDelete(syncSemId); taskDelete(tidtask1); }

# 1. 介绍 ## 1.1 什么是Knative Knative 是一个开放式、云原生的服务编排和可伸缩的平台，由 Google、红帽、IBM 和 SAP 共同推出。它为构建、部署和管理现代、轻量级的无服务器应用程序提供了一个开放式的、可扩展的平台。 Knative 提供了三个核心组件：构建（Build）、服务（Serving）和事件（Eventing）。构建组件用于构建和打包应用程序镜像，服务组件用于部署和管理应用程序服务，而事件组件则用于实现事件驱动的架构，通过消息队列进行事件的传递和处理。 ## 1.2 消息队列的作用和重要性消息队列是一种常用的分布式系统间通信的机制，通过将消息存储在队列中，实现了异步通信的能力。当一个系统产生一个消息时，不需要立即处理，而是将消息发送到消息队列中，待其他系统空闲时再进行处理。这样可以实现系统间的解耦和异步处理，提升系统的可扩展性、可靠性和可伸缩性。消息队列在现代应用开发中扮演着重要的角色，特别是在微服务架构和分布式系统中。它能够将不同服务之间的通信进行解耦，提供高可靠性的消息传递机制，支持通过异步方式处理大量的任务和事件。消息队列对于实现事件驱动的架构也是至关重要的。通过消息队列，系统可以将事件发送给感兴趣的订阅者，订阅者只需要关心自己感兴趣的事件，而不需要关心事件是如何产生和传递的。这样可以实现松耦合的系统架构，支持快速的业务变更和扩展。综上所述，消息队列在现代应用开发中具有重要的作用和重要性。在 Knative 中，与消息队列的集成可以帮助开发者更好地实现基于事件驱动的架构，提升应用程序的可靠性和可扩展性。 # 2. 消息队列基础消息队列是一种用于在组件间传递消息的通信机制。它具有以下特点： #### 2.1 消息队列的概念和原理消息队列通过提供异步通信机制，将消息从发送者传递到一个或多个接收者。它通常基于生产者-消费者模式，其中生产者将消息推送到队列，而消费者则从队列中拉取消息。消息队列能够解耦组件之间的通信，提高系统的可伸缩性和弹性。消息队列的核心原理包括消息传递、队列存储和消息确认机制。消息传递指消息从发送者传递到接收者的过程；队列存储指消息在队列中的存储方式，通常采用先入先出的方式；消息确认机制则确保消息在传递过程中不会丢失，包括消息确认、消息重发等。 #### 2.2 常见的消息队列解决方案常见的消息队列解决方案包括 RabbitMQ、Kafka、ActiveMQ 等。这些解决方案针对不同的场景和需求提供了多样化的功能和特性，如可靠性消息传递、高吞吐量、持久化存储、消息分区等。在实际应用中，选择合适的消息队列解决方案需要考虑系统的可靠性要求、吞吐量需求、消息延迟等因素。同时也可以根据特定业务场景进行定制开发，提供定制化的消息队列功能。以上是消息队列基础的介绍，接下来将详细介绍Knative中的消息队列集成。 # 3. Knative中的消息队列集成 Knative作为一个开放、可扩展的平台，与消息队列的集成可以提供更强大的事件驱动能力。在这一章节中，我们将介绍Knative与消息队列的关系以及Knative消息队列集成的优势和局限性。 #### 3.1 Knative与消息队列的关系 Knative与消息队列之间存在紧密的关系。消息队列是一种常见的解决方案，用于解耦应用程序的不同部分。它允许发送者将消息发送到队列中，而接收者则可以从队列中接收和处理这些消息。消息队列的主要作用是实现异步通信，使不同组件之间的通信更加灵活、可靠和可伸缩。 Knative通过集成各种消息队列解决方案，如Kafka、RabbitMQ等，为开发者提供了在应用程序中使用消息队列的能力。Knative通过自动化管理和扩展这些消息队列，提供了可靠的事件驱动能力。 #### 3.2 Knative消息队列集成的优势与局限性 Knative消息队列集成的优势有以下几点： - **解耦和灵活性**：Knative与消息队列的集成可以实现不同部分之间的解耦，使得应用程序的不同组件可以独立进行开发、部署和扩展。通过将消息发送到队列中，接收者可以根据自己的需求和处理能力自由选择何时以及如何处理这些消息。 - **可靠性和可伸缩性**：Knative自动管理和扩展消息队列，确保消息的可靠传递和处理。无论消息队列所需的吞吐量是多少，Knative都可以自动进行扩展，以满足应用程序的需求。这使得应用程序在面对高峰期时能够保持高可用性和性能。 Knative

最低0.47元/天解锁专栏

买1年送3月

点击查看下一篇

百万级高质量VIP文章无限畅学

千万级优质资源任意下载

C知道免费提问 ( 生成式Al产品 )

Knative 中的消息队列集成和事件处理实践

相关推荐

专栏目录

专栏目录

Knative 中的消息队列集成和事件处理实践

相关推荐

利用消息队列实现进程间通信

Java中实现消息队列的策略与代码实践

全栈Serverless 应用开发指南与实践.zip

无服务架构实践手册（Serverless Handbook）.pdf

深入理解Serverless开发者实践指南

阿里巴巴云原生技术与架构实践精华资料.zip

Knative 中的自动扩展和自动缩减：动态资源管理技术解析

Kubernetes_K8s中的Serverless架构与Knative实践

Knative 与微服务架构：从概念到实践

专栏目录

最新推荐

【AXWReport控件快速入门】：报表生成的高效路径

信令捕获工具精挑细选指南：打造您的5G分析利器

微信小程序tabbar状态管理与性能优化：心跳报文的深度解析与最佳实践

【安川机器人GP7_8全方位速成课程】：零基础快速掌握操作与编程

【bq24195深度剖析】

niRvana 4.5.3：手机端博客用户体验提升全攻略

【SAS统计分析精通指南】：解锁数据真相的7大基础概念与操作秘诀

【CCS5.5终极指南】：全方位精通CCS5.5，从安装到高级调试技术

CASIC协议深度剖析：导航接收机技术的核心要点揭露

专栏目录