有个程序的某个事件会不断实时产生大量数据，产生的数据包含多个机器的数据，但是要对数据进行分类，比如判断数据是1(机器编号)的数据则写入1.log，总共有一百台机器。遇到的问题是当程序只写入20台机器的数据时，每个log文件里记录的数据是实时的，而随着机器数量的增加，log文件里的记录的数据就会有延迟(记录的是好几分钟前的数据，而不是现在的产生的数据)。请问如何程序如何设计才能使一百多个log文件记录的数据是实时的？
我写了一个单机测试demo，模拟100个生产者情况，bus选型是rx.net
nuget:System.Reactive
模拟代码
public class Program
{
//设计目标，不确定数量的生产者高速产生数据（其中个别生产者生产频率超高）
//分门别类将数据记录到对应的log.txt中
static Subject<MsgPackage> bus = new Subject<MsgPackage>();
static void Main(string[] args)
{
//模拟demo，采用rx作为bus。
//暂时不考虑复杂的降权算法，我们先做一种简单策略，先用一个总体订阅判定缓存，缓存没有新建单独订阅任务，缓存有更新缓存生存期
//编写demo，我就暂时不引入缓存系统了，直接使用字典做了，如果你有兴趣用缓存系统，可以直接引入
//原本考虑使用valuetask，不过不想给demo引入过多的技术特征，还是选择了直接使用IDisposable
//System.Collections.Concurrent.ConcurrentDictionary
// <int, ValueTask> dic = new ConcurrentDictionary<int, ValueTask>();
System.Collections.Concurrent.ConcurrentDictionary
<int, IDisposable> dic = new ConcurrentDictionary<int, IDisposable>();
bus.Subscribe(p =>
{
//每一个数据源分开订阅处理，这样就不会因为个别生产者捣乱，让其他生产者数据处理不即使
if (!dic.ContainsKey(p.id))
{
dic.GetOrAdd(p.id, msgHandler(p.id));
}
});
//模拟100个生产者
for (int i = 0; i < 100; i++)
{
var j = i;
Task.Run(async () =>
{
var n = j + 1;
while (true)
{
bus.OnNext(new MsgPackage()
{
id = j,
playload = DateTime.Now.ToString()
});
await Task.Delay(n * 100); //根据j延时，让生产者生产消息，很明显这里有些生产的快，有些生产的慢
}
});
}
Console.ReadKey();
}
public static IDisposable msgHandler(int key)
{
string filename = $"{key}.log";
return bus.Where(p => p.id == key).Subscribe(msg => {
var lines=new string[] { $"{msg.id}\t{msg.playload}"};
File.AppendAllLines(filename, lines);
});
}
/// <summary>
/// 消息对象
/// </summary>
public class MsgPackage
{
public int id { get; set; }
public string playload { get; set; }
}
}
你其实没必要将所有机器上的日志都由一台机器收集，完全可以 每台机器自己管理自己的日志，管理机想要获取某台日志的时候，再将log的管理权拿过来（也就是在管理机这边实现log的缓存）
我的理解是 你不会同时打开200台机器的日志看，需要看哪台就实时获取哪台的日志。不需要读的，就本地缓存
硬盘慢是没办法的，除非你文件本身分散在多个硬盘上，否则多线程也解决不了这个问题。
你只有描述，没有任何代码，所以我无法判定你现在遇到的是什么问题。我只能说，如果是我我会怎么写代码
先说单机方式：
1.先将数据写入一个总线---------你可以用eventbus，Rx，或者就是nlog，log4net本身（当然，数据本身需要有特征，就像你说你要判定区分，如果数据没特征还怎么区分）
2. 如果已经在总线里，那么只需在总线里挑选数据（如果是log4net你可以写个过滤器）
比如如果我用rx，他的结果是
rx.where（p=>p.name==“A”).订阅(p=>写入文件) /////////从数据总线里过滤A机器的数据，然后写入文件
在说多机方式：
1.多机和单机其实没区别。大体上依旧还是需要一个总线，这个总线可以是mqtt，Azure bus，kafafa，rabiitmq，activemq
比如我们可以用mqtt
假设我机器A写日志的时候这么写：
topic：日志记录/机器A
playload：{时间，记录}----这是json，我理解就好，我不必写完整格式
那么如果非要分类写文本log，我们可以起一个任务，订阅这个总线
mqtt.订阅（“日志记录/#”).处理日志(p=>{
string name=p.topic.split("/")[1]; 同样这里你理解意思就好，我把topic处理一下，就可以知道消息是谁发的
//根据name写playload到具体文件
})
ps：如果新做项目，这是就是目前新做项目的手段。
如果是老项目，那就没啥办法了，那的是采集了在弄。这个啥一堆的java 分布式日志采集方案，我就不赘述了，（甚至还包括hadoop的分布式文件系统mapreduce都能算你问题的方案）
总体上，传统分布式采集，分布式文件，那是旧系统兼容方案。而新系统直接上bus就行（甚至你直接上dapper做log，上fink，上普罗米修斯做log我也不反对）