1.RocketMQ原理（4）——消息ACK机制及消费进度管理
2.c++大学生个人消费管理系统 大神救命！消费消费！管理管理急用！源码源码！消费消费！管理管理！源码源码虚拟源码网站
3.RocketMQ 消费者（2）客户端设计和启动流程详解 & 源码解析
4.开源项目轻量元数据管理解决方案——Marquez
5.Kafka消费者源码：重平衡（1）-初始化与FIND_COORDINATOR
RocketMQ原理（4）——消息ACK机制及消费进度管理
在 RocketMQ 中，消费消费消息的管理管理 ACK 机制和消费进度管理是保证消息成功消费的关键。在 PushConsumer 中，源码源码消息消费的消费消费管理主要通过消费回调来实现。当业务实现消费回调时，管理管理只有在回调函数返回 ConsumeConcurrentlyStatus.CONSUME_SUCCESS 的源码源码情况下，RocketMQ 才会认为该批消息（默认每批为 1 条）已被成功消费。消费消费如果消息消费失败，管理管理例如遇到数据库异常或余额不足等情况，源码源码业务应返回 ConsumeConcurrentlyStatus.RECONSUME_LATER，表示消息需要重新尝试。
为了确保消息至少被成功消费一次，RocketMQ 会将消费失败的有源码怎么搭建网站源码消息重新投递给 Broker（消息主题将变更为重试主题），并在指定时间（默认为 秒，可配置）后再次将消息投递到该 ConsumerGroup。如果消息在多次尝试后仍无法成功消费，则会投递到死信队列，应用程序可以监控死信队列并采取人工干预措施。
当启动一个新的实例时，PushConsumer 会根据先前存储的消费进度（consumer offset）来发起第一次 Pull 请求。如果当前消费进度在 Broker 中不存在，这表明是一个全新的消费组，此时客户端可以选择不同策略。社区中常见的一种疑问是：“为什么我设置了 CONSUME_FROM_LAST_OFFSET，但历史消息还是被消费了？” 这是因为只有全新的消费组才会使用特定策略，而老的消费组则会继续按已存储的进度消费。
为了优化性能并减少重复消费的风险，RocketMQ 采用一种与单条消息单独 ACK 不同的机制来管理消费进度。消费进度记录的是批次中最小的 offset 值，这意味着如果一批消息中有多个 offset，只有最小的社工库源码搜云社工库源码 offset 会被更新。这种设计可以提高性能，但也带来潜在的重复消费问题，即消费进度可能仅更新至已消费消息的最小 offset，导致后续消息被重复消费。为解决这一问题，RocketMQ 在较新版本中引入了流控机制，通过配置 consumeConcurrentlyMaxSpan，当缓存中消息的最大值与最小值差距超过此阈值（默认为 ）时，会暂停消息的拉取，以缓解重复消费风险。
尽管如此，解决消费进度卡住的问题，最直接的方法是设置消费超时时间。在 RocketMQ 3.5.8 及之后的版本中，引入了超时处理机制，以应对消费进度卡住的情况。通过源码分析，可以看到该方案在一定程度上解决了消费进度卡住的提供源码和不提供源码价格区别问题，但仍存在一些不足之处。
c++大学生个人消费管理系统 大神救命！！急用！！！！
工大课设吧，我这里有，你看看有没有帮助，，很多呀，我要怎么给你，给你分享吧。。
直接给你吧。。。即时通讯im源码红包源码。。。希望对你有所帮助！！！！！
#include <stdio.h>
#include <stdlib.h>
#define FilePath1 "Myinfor.dat"
#define FilePath2 "Myinfor.txt"
#define Status int
#define OK 1
#define Error 0
#define NotFound 2
typedef struct Infor{
int month;
int spxf;
int fz;
int znjy;
int sdf;
int ylf;
int cx;
int byzhf;
} Infor,*Infor1;
typedef struct pType{
int no;
int data;
}pType;
void menu(void);
void input1(Infor *newI,int mon);
void input(Infor *newI);
void writeinfor(Infor *newI);
void changeFormat(void );
Status search(Infor *a,int mon);
void paixu(Infor *a);
void modify(Infor *a,int mon);
void delRecord(int mon);
void xuanze(int item);
void xiugai(int m);
int panduan(Infor *a,int mon);
void main()
{ while(1)
{ menu(); }
}
void menu(void)/*菜单*/
{ int item;
printf("\n………\"我的大学\"生活消费管理系统…………\n\n");
printf("\t\t1.…………录 入 数 据………….\n");
printf("\t\t2.…………查 看 数 据………….\n");
printf("\t\t3.…………修 改 数 据………….\n");
printf("\t\t4.…………查 询 数 据………….\n");
printf("\t\t5.…………排 序 数 据………….\n");
printf("\t\t6.…………删 除 数 据………….\n");
printf("\t\t0.…………退 出 系 统………….\n");
printf("请输入要进行的操作: " );
scanf("%d",&item);
if(item>6 || item<-1)
{ printf("请重新输入要进行的操作: " );
menu(); }
else xuanze( item); }
int panduan(Infor *a,int mon)
{ int item;
FILE *fp;
fp=fopen(FilePath1,"ab+");
if(fp==NULL)
{ printf("无法创建文件：%s",FilePath1);
exit(0); }
if(mon<=)
{ item=search(a,mon);
while(item==OK)
{ printf("输入月份已存在请重新输入要建立的月份:\n");
scanf("%d",&mon);
item=search(a,mon); } }
else{
printf("您输入的月份有误请重新输入:\n");
scanf("%d",&mon);
panduan(a,mon); }
fclose(fp);
return mon; }
void xuanze(int item)
{ int mon;
Infor *a;
a=(Infor *)malloc(sizeof(Infor));
switch(item)
{ case 0: //getchar();/*退出*/
//getchar();
printf("\n ……………………欢迎使用…………………………");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t. \"我的大学\"生活消费管理系统 .");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t. 欢迎下次使用 .");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
printf(" \n\t.\t\t\t\t\t\t.");
//printf("\n\"我的大学\"生活消费管理系统\n\n\n\n");
//printf("\t\t\t\t\t\n. 欢迎下次使用 \n\n\n\n");
printf("\n……………………………………………………………\n\n\n\n");
exit(1);
break;
case 1:
printf("请输入要建立的月份:\n");
scanf("%d",&mon);
mon=panduan(a,mon);
input1(a,mon);
writeinfor(a);
break;
case 2:
changeFormat();
break;
case 3:
printf("请输入要查找的月份:\n");
scanf("%d",&mon);
item=search(a,mon);
mon=a->month;
if (item!=OK) printf("\n没有符合条件的记录!\n");
else
{
printf("\n 记录月份 食品消费 房租 子女教育费用 水电费 医疗费 储蓄 本月总花费 \n");
printf("------------------------------------------------------- \n");
printf("%7d %8d %8d %8d %8d %8d %8d %8d\n",a->month,a->spxf,a->fz,a->znjy,a->sdf,a->ylf,a->cx,a->byzhf);
input(a);
modify(a,mon); }
break;
case 4:
printf("请输入要查找的月份:\n");
scanf("%d",&mon);
item=search(a,mon);
if (item!=OK) printf("\n没有符合条件的记录!\n");
else{
printf("\n记录月份 食品消费 房租 子女教育费用 水电费 医疗费 储蓄 本月总花费 \n");
printf("------------------------------------------------------- \n");
printf("%7d %8d %8d %8d %8d %8d %8d %8d\n",a->month,a->spxf,a->fz,a->znjy,a->sdf,a->ylf,a->cx,a->byzhf);
}
break;
case 5:
printf("请输入要查找的月份:\n");
scanf("%d",&mon);
item=search(a,mon);
if (item!=OK) printf("\n没有符合条件的记录!\n");
else
paixu(a);
break;
case 6:
printf("请输入要查找的月份:\n");
scanf("%d",&mon);
item=search(a,mon);
mon=a->month;
if (item!=OK) printf("\n没有符合条件的记录!\n");
else
{
printf("------------------------------------------------------- \n");
printf("%7d %8d %8d %8d %8d %8d %8d %8d\n",a->month,a->spxf,a->fz,a->znjy,a->sdf,a->ylf,a->cx,a->byzhf);
delRecord(mon); }
break;}
free(a);}
void input1(Infor *newI,int mon)
{ printf("\n请依次输入数据[说明:中间以空格符隔开]:\n( 食品消费 房租 子女教育费用 水电费 医疗费 储蓄 )\n");
scanf("%d%d%d%d%d%d",&newI->spxf,&newI->fz,&newI->znjy,&newI->sdf,&newI->ylf,&newI->cx);
newI->month=mon;
newI->byzhf=newI->spxf+newI->fz+newI->znjy+newI->sdf+newI->ylf+newI->cx;
fflush(stdin);}
void input(Infor *newI)
{
printf("\n请依次输入数据[说明:中间以空格符隔开]:\n( 食品消费 房租 子女教育费用 水电费 医疗费 储蓄 )\n");
scanf("%d%d%d%d%d%d",&newI->month,&newI->spxf,&newI->fz,&newI->znjy,&newI->sdf,&newI->ylf,&newI->cx);
newI->byzhf=newI->spxf+newI->fz+newI->znjy+newI->sdf+newI->ylf;
fflush(stdin);}
void writeinfor(Infor *newI)
{
FILE *fp;
fp=fopen(FilePath1,"ab+");
if(fp==NULL)
{ printf("无法创建文件：%s",FilePath1);
exit(0);}
fwrite(newI,sizeof(Infor),1,fp);
fclose(fp);
printf("数据录入成功！\n");}
void changeFormat(void)
{
FILE *fp1,*fp2;
Infor *a;
a=(Infor *)malloc(sizeof(Infor));
fp1=fopen(FilePath1,"rb+");
if(fp1==NULL)
{ printf("无法找到文件：%s\n",FilePath1);
return ;}
fp2=fopen(FilePath2,"wt+");
if(fp2==NULL)
{ printf("无法创建文件：%s\n",FilePath2);
return ;
}
//fputs(" \n!@#￥%……&*（&……￥#@@?\"我的大学\"生活消费管理系统！@#￥%……&*（&……￥#@@!n\n",fp2);
fputs("记录月份 食品消费 房租 子女教育费用 水电费 医疗费 储蓄 本月总花费 \n",fp2);
fputs("---------------------------------------------------- \n",fp2);
printf("\n记录月份 食品消费 房租 子女教育费用 水电费 医疗费 储蓄 本月总花费 \n");
printf("--------------------------------------------------- \n");
rewind(fp1);
fread(a,sizeof(Infor),1,fp1);
while(!feof(fp1))
{ printf("%7d %8d %8d %8d %8d %8d %8d %8d\n",a->month,a->spxf,a->fz,a->znjy,a->sdf,a->ylf,a->cx,a->byzhf);
fprintf(fp2,"%7d %8d %8d %8d %8d %8d %8d %8d\n",a->month,a->spxf,a->fz,a->znjy,a->sdf,a->ylf,a->cx,a->byzhf);
fread(a,sizeof(Infor),1,fp1);}
fputs("--------------------------------------------------- \n",fp2);
fputs("关闭本程序继续原程序！\n",fp2);
fclose(fp1);
fclose(fp2);
system(FilePath2);
remove(FilePath2);}
Status search(Infor *a,int mon)
{
FILE *fp1;
int isfound=0;
//printf("请正确输入要查询的月份：");
//scanf("%d",&mon);
fflush(stdin);
fp1=fopen(FilePath1,"rb+");
if(fp1==NULL)
{ printf("无法找到文件：%s\n",FilePath1);
return Error; }
rewind(fp1);
fread(a,sizeof(Infor),1,fp1);
while(!feof(fp1))
{ if(a->month==mon)
{ isfound=1;
Break; }
else
{ isfound=0; }
fread(a,sizeof(Infor),1,fp1);}
fclose(fp1);
if(isfound)
return OK;
else
return NotFound; }
void paixu(Infor *a)
{ int i=0,j=0,flag=0,t;
pType px[8]={ { 0,0}};
char str[8][]={ "记录月份","食品消费"," 房租", "子女教育费用", "水电费"," 医疗费"," 储蓄"," 本月总花费"};
for(;i<8;i++)
px[i].no=i;
px[0].data=a->month;
px[1].data=a->spxf;
px[2].data=a->fz;
px[3].data=a->znjy;
px[4].data=a->sdf;
px[5].data=a->ylf;
px[6].data=a->cx;
px[7].data=a->byzhf;
for(i=1;i<8;i++)
{
flag=0;
for(j=0;j<8-i;j++)
if(px[j].data>px[j+1].data)
{ t=px[j].data;
px[j].data=px[j+1].data;
px[j+1].data=t;
t=px[j].no;
px[j].no=px[j+1].no;
px[j+1].no=t;
flag=1; }
if(flag==0) break;}
printf("\n");
for(i=0;i<8;i++)
{ printf(" %s",str[px[i].no]);}
printf("\n----------------------------------------------------- \n");
for(i=0;i<8;i++)
{ printf("%8d ",px[i].data); }
printf("\n");}
void modify(Infor *a,int mon)
{ FILE *fp1,*fp2;
Infor *b;
b=(Infor *)malloc(sizeof(Infor));
fp1=fopen(FilePath1,"rt");
fp2=fopen("temp.dat","wt+");
rewind(fp1);
fread(b,sizeof(Infor),1,fp1);
while (!feof(fp1))
{ if(b->month==mon)
{
fwrite(a,sizeof(Infor),1,fp2);
}
else
{ fwrite(b,sizeof(Infor),1,fp2);
}
fread(b,sizeof(Infor),1,fp1);
}
fclose(fp1);
fclose(fp2);
remove(FilePath1);
rename("temp.dat",FilePath1);
printf("修改数据成功！\n" );
changeFormat();
}
void delRecord(int mon)
{
FILE *fp1,*fp2;
Infor *b;
b=(Infor *)malloc(sizeof(Infor));
fp1=fopen(FilePath1,"rt");
fp2=fopen("temp.dat","wt+");
rewind(fp1);
fread(b,sizeof(Infor),1,fp1);
while (!feof(fp1))
{
if(b->month!=mon)
fwrite(b,sizeof(Infor),1,fp2);
fread(b,sizeof(Infor),1,fp1);
}
fclose(fp1);
fclose(fp2);
remove(FilePath1);
rename("temp.dat",FilePath1);
printf("删除数据成功！\n" );
changeFormat();
}
RocketMQ 消费者（2）客户端设计和启动流程详解 & 源码解析
RocketMQ 消费者系列的第二篇文章深入剖析了客户端设计和启动流程。本文将带你了解消费者类的结构、启动过程，以及源码细节。
首先，消费者客户端设计的核心是DefaultMQPullConsumer和DefaultMQPushConsumer，它们都实现了消费者接口，并扩展了客户端配置类。DefaultXXXXConsumer实际上是一个代理，内部通过DefaultMQXXXXConsumerImpl执行大部分方法，后者包含了MQClientInstance，它是客户端实例的管理核心，负责与Broker通信和存储元数据。
消费者启动涉及这三个关键类：DefaultMQPullConsumer/ConsumerImpl和MQClientInstance。启动流程分为新建消费者、消费者启动以及客户端实例的初始化。拉消费者和推消费者虽然操作不同，但内部都依赖拉取消息服务，如PullMessageService，推消费者还利用ConsumeMessageService接口进行并发或顺序消费。
拉模式和推模式的消费者启动流程相似，但推消费者更注重消息推送的自动处理。在DefaultMQPushConsumer的启动中，实际是调用其代理类的启动方法，而MQClientInstance则负责初始化客户端通信和设置。
源码解析部分，我们会在后续文章中详细剖析DefaultMQProducerImpl和MQClientInstance的启动过程。想要获取更多消息中间件的源码解析和最新动态，别忘了关注我们的公众号消息中间件(middleware-mq)，同时，本文由OpenWrite平台发布。
开源项目轻量元数据管理解决方案——Marquez
轻量级元数据管理解决方案——Marquez
Marquez，由WeWork开源的元数据管理工具，专为简化数据生态系统元数据的收集、聚合和可视化而设计。它提供了一个轻量级的元数据服务，帮助用户全面掌握数据集的产生和消费情况，以及数据处理过程的可视化，并集中管理数据集的生命周期。
Marquez在持续发展中，当前标星数为1.5K，最新版本发布于三周前的0..1，主要使用Java和TS语言开发。部署方式与Java项目类似，只需启动对应Web端服务和API服务。Marquez的血缘API简洁高效，便于建立数据血缘依赖关系，确保数据分析质量。如需获取安装包、源代码及学习资料，可访问官网或使用大数据流动后台回复“Marquez”。
Marquez的安装流程简洁，通过命令行即可快速完成。启动命令如下：$ git clone github.com/MarquezProject/marquez && cd marquez$ ./docker/up.sh --seed，之后通过访问/OpenLineage/...", "schemaURL": "openlineage.io/spec/1-0..." }' 完成任务后，使用类似代码进行：$ curl -X POST /OpenLineage/...", "_schemaURL": "github.com/OpenLineage/...", "fields": [ { "name": "a", "type": "VARCHAR"}, { "name": "b", "type": "VARCHAR"} ] } } }], "producer": "github.com/OpenLineage/...", "schemaURL": "openlineage.io/spec/1-0..." }' 正常运行应接收到 CREATED的响应，并在页面上找到血缘展示。
Marquez不仅简化了元数据管理，还提供了标准的元数据采集方案，目前支持Spark、Airflow的表级别和列级别数据血缘收集，而Flink仅支持表级别的血缘收集。Marquez未来有望支持更多数据源，共同期待其发展。
Kafka消费者源码：重平衡（1）-初始化与FIND_COORDINATOR
在Kafka 2.5.2的消费者组中，重平衡是关键，它定义了消费者如何根据订阅关系调整对Topic分区的分配。当消费者数量、订阅的Topic或GroupCoordinator所在的Broker发生变更时，会触发重平衡。
消费者组状态由GroupState类管理，共有五个状态：Empty（无成员）、PreparingRebalance（加入中）、CompletingRebalance（等待分配）、Stable（已平衡）和Dead（元数据已删除）。状态间的转换基于预先定义的前置状态。例如，从Empty到PreparingRebalance，预示着重平衡的开始。
重平衡过程分为几个步骤，首先是消费者和Broker之间的协调。服务端启动时，GroupCoordinator组件即已就绪，而Consumer通过ConsumerCoordinator与之通信。在启动时，消费者首先会通过FindCoordinatorRequest找到GroupCoordinator，通过最小负载节点发送请求，然后服务端确定哪个Broker负责协调，如groupId的hash值对consumer_offsets分区数取模确定。
一旦找到GroupCoordinator，消费者会发送JoinGroupRequest。后续步骤如SYNC_GROUP和HEARTBEAT确保消费者与协调器保持同步。这部分详细内容在后续的文章中会进一步探讨。
