SRS定位是运营级的互联网直播服务器集群，追求更好的概念完整性和最简单实现的代码。
运营级：商业运营追求极高的稳定性，良好的系统对接，以及错误排查和处理机制。譬如日志文件格式，reload，系统HTTP接口，提供init.d脚本，转发，转码，边缘回多源站，都是根据CDN运营经验作为判断这些功能作为核心的依据。
互联网：互联网最大的特征是变化，唯一不变的就是不断变化的客户要求，唯一不变的是基础结构的概念完整性和简洁性。互联网还意味着参与性，听取用户的需求和变更，持续改进和维护。
直播服务器：直播和点播这两种截然不同的业务类型，导致架构和目标完全不一致，从运营的设备组，应对的挑战都完全不同。两种都支持只能说明没有重心，或者低估了代价。
集群：FMS(AMS)的集群还是很不错的，虽然在运营容错很差。SRS支持完善的直播集群，Vhost分为源站和边缘，容错支持多源站切换、测速、可追溯日志等。
概念完整性：虽然代码甚至结构都在变化，但是结构的概念完整性是一直追求的目标。从SRS服务器，P2P，ARM监控产业，MIPS路由器，服务器监控管理，ARM智能手机，SRS的规模不再是一个服务器而已。
简单实现：对于过于复杂的实现，宁可不加入这个功能，也不牺牲前面提到的要求。对于已经实现的功能的代码，总会在一个版本release前给予充分的时间来找出最简答案。不求最高性能，最优雅，最牛逼，但求最简单易懂。
RTMP协议详解：http://blog.sina.com.cn/s/blog_541348370102y3me.html
                            
http://blog.csdn.net/weixin_39799839/article/details/78765972


SRS git地址:https://gitlab.com/winlinvip/srs-gitlab
SRS
 代码分析
1.SRS 代码分析【服务器启动】
2.SRS 代码分析【RTMP连接请求响应】
3.SRS 代码分析【转发流实现】
4.SRS 代码分析【RTMP信息play/publish】
5.SRS 代码分析【FLV文件解析】
6.SRS 代码分析【HTTP-FLV传输实现】
7.SRS 代码分析【HLS切片】
8.rtmp流转Hls流 与 flv转mpegts
9.SRS 代码分析【mpeg-ts解析】
10.SRS 代码分析【DVR录像实现】
11.SRS 代码分析【RTMP握手实现】
12.SRS 代码分析【RTMP Chunck数据读取】
13.SRS 代码分析【RTMP Chunck数据发送】
14.RTMP协议中的AMF数据类型说明
15.SRS 代码分析【保存AAC音频文件】
16.SRS 代码分析【保存MP3音频文件】

目录：

《Kafka Producer设计分析》

《KafkaProducer类代码分析》

《RecordAccumulator类代码分析》

《Sender类代码分析》

《NetworkClient类代码分析》

-------------------------------------------------------------------

<上一节《你绝对能看懂的Kafka源代码分析-Kafka Producer设计分析》

上一小节，我们从设计角度对producer进行了讲解，我们也从宏观上了解了producer的设计原理，接下来我们将进入代码的分析

----------------------------------------------------------------------

客户端发送消息时调用KafkaProducer的send方法，所以我们先分析KafkaProducer，再层层深入。

KafkaProducer相当于整个快递公司的总控员，操作收件员收件，命令分拣员进行分拣，最终通知货车可以发车了。货车发货则在另外一个线程中进行。

消息发送的顶层逻辑都在KafkaProducer中。在看代码前，我先宏观介绍下KafkaProducer中send方法的主流程，帮助代码理解，括号中以发快递类比：

拦截器处理（对快递进行发送前的预处理）
	
判断producer是否可用。依据是负责io的线程是否工作。（车队罢工了还如何发送快递？）
	
判断metadata是否可用。metadata相当于调度员的指挥图，存储了kafka集群的各种信息，包括topic、分区等等。（如果没有快递网点的信息，如何进行调度派发？）
	
对key和value进行序列化。（对快递进行包装）
	
获取要发往的分区编号（快递目的地部分地址）
	
计算序列化后大小（快件称重）
	
通过RecordAccumulator把消息加入batch中（分拣员进行分拣）
	
如果batch满了，或者创建了新batch来容纳消息，则唤醒sender线程执行发送。（已经有封箱的货物了，发货吧！）
send方法的整体逻辑如上，每一步其实都和我们真实场景相对应，这就是设计的巧妙之处。

接下来我们进入代码部分，我们先看下KafkaProducer中的部分重要属性

private final Partitioner partitioner;
说明：分区选择器，根据一定策略，选择出消息要发往的分区
private final int maxRequestSize;
说明：消息最大程度，包括了消息头、序列化后的key和value
private final long totalMemorySize;
说明：单个消息发送的缓冲区大小
private final Metadata metadata;
说明：kafka集群元数据
private final RecordAccumulator accumulator;
说明：前面所说的分拣员，维护不同分区的batch，负责分拣消息。等待sender来发送
private final Sender sender;
说明：发送消息，实现Runable，ioThread中启动运行
private final Thread ioThread;
说明：运行sender的线程，负责发送消息到kafka集群
private final CompressionType compressionType;
说明：压缩算法类型，gzip等。消息数量越多，压缩效果越好
private final Serializer<K> keySerializer;
说明：key的序列化器
private final Serializer<V> valueSerializer;
说明：value的序列化器
private final ProducerConfig producerConfig;
说明：生产者的配置
private final long maxBlockTimeMs;
说明：等待更新kafka集群元数据的最大时长
private final ProducerInterceptors<K, V> interceptors;
说明：发送前消息钱，要先通过一组拦截器的处理。也可以先于用户的callback预处理

KafkaProducer构造器会初始化上面的属性。另外在构造函数最后可以看到启动了ioThread。

this.sender = newSender(logContext, kafkaClient, this.metadata);
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
this.ioThread.start();

接下来我们看一下send方法代码：

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // intercept the record, which can be potentially modified; this method does not throw exceptions
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    return doSend(interceptedRecord, callback);
}

ProducerInterceptors对象中维护了List<ProducerInterceptor<K, V>> interceptors。在onSend方法中，会循环这个列表，调用每个ProducerInterceptor的onSend方法做预处理。

拦截器处理后，再调用doSend方法。发送的主要逻辑都在doSend方法中，我按照上面介绍的发送主流程结合代码来讲解。这里不粘贴doSend方法的整段代码，大家自行参考源代码。

1、判断producer是否可用。

可以看到一进来就调用了throwIfProducerClosed()方法，这个方法里逻辑如下：

private void throwIfProducerClosed() {
    if (ioThread == null || !ioThread.isAlive())
        throw new IllegalStateException("Cannot perform operation after producer has been closed");
}

很简单，就是在检查io线程状态。

2、判断这个topic的metadata是否可用。

代码如下：

ClusterAndWaitTime clusterAndWaitTime;
try {
    clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
} catch (KafkaException e) {
    if (metadata.isClosed())
        throw new KafkaException("Producer closed while send in progress", e);
    throw e;
}

主要逻辑在waitOnMetadata方法中。这里不展开讲，方法中做的事情是从缓存获取元数据中topic的信息，判断是否可用，如果缓存中存在分区并且请求分区在范围内，则直接返回cluster信息，否则发送更新请求。再判断分区是否正常并且请求更新的分区在此topic分区的范围内。这个方法的返回值是Cluster原数据以及所花费的时长。

3、对key和value进行序列化。

主要是如下两行代码

serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());

3、获取消息要发往的分区编号

int partition = partition(record, serializedKey, serializedValue, cluster);
tp = new TopicPartition(record.topic(), partition);

TopicPartition对象实际上只是封装了topic和partition，那么消息的发送地址就齐全了。

4、计算序列化后大小

int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),
        compressionType, serializedKey, serializedValue, headers);
ensureValidRecordSize(serializedSize);


ensureValidRecordSize方法中验证size是否未超过maxRequestSize及totalMemorySize。

5、通过RecordAccumulator把消息加入batch中

RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
        serializedValue, headers, interceptCallback, remainingWaitMs);

accumulator.append方法中做分拣逻辑处理，后面会重点讲解RecordAccumulator。这里我们只需要知道通过这个方法处理，你的消息已经缓存到待发送Batch中。

6、如果batch正好满了，或者创建了新batch来容纳消息，则唤醒sender线程执行发送。

if (result.batchIsFull || result.newBatchCreated) {
    log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
    this.sender.wakeup();
}

至此Producer中send方法的主要代码逻辑已经分析完毕。下一小节我们看一下分拣员RecordAccumulator是如何实现的。

下一节《RecordAccumulator类代码分析》

-------------------------------------------------------------------------------------------------

附KafkaProducer类部分注释翻译：

producer 线程安全，跨线程共享一个producer对象，通常会比多个对象更快。

producer由一个buffer空间池组成，它容纳了没有被传输到server的数据。同时有一个后台 I/O线程负责把这些数据记录转化为reqeust，然后把他们发送给集群。如果producer用后未成功关闭，这些资源将被泄漏。

send()方法是异步的。当调用他添加记录到等待发送的数据缓冲区会立即得到返回。这允许producer把独立的消息打包起来，这样会更为高效。

ack配置项控制认为请求完成的条件。设置为“all”，数据全部提交前，是不会返回结果的，这是最慢，但是持久性最好的。

如果request失败，producer会自动重试，但是如果我们设置了retries为0，则不会重试。开启重试，会出现数据重复的可能性。

producer维护了每个partition对于未发送消息的缓冲区。缓冲区的大小通过batch.size配置项指定。设置的大一点，可以让batch更大，但是也需要更多的内存。

默认的，即使buffer还有未使用的空间，也是可以被立即发送出去的。然而，如果你想减少请求的次数，你可以设置linger.ms为大于0的值。这会让producer等待相应的时间，以让更多的数据到达batch后再发送。这个和Nagle在TCP里的算法是类似得。例如上面的代码片段，因为我们设置了linger为1ms，可能100条记录会在一次请求中发送出去。你也可以再加1ms，让请求等待更多的数据到达，如果我们的bufer还没填满的话。注意，相近时间到达的数据通常会在同一个batch中，即使linger.ms=0。所以，由于不计后果的linger设置，将会导致出现过重的数据负载batch。所以当你负载并不重，如果把linger设为大于0，会让请求更少，更高效，这只会造成很小的延迟。

buffer.memory控制producer可用的全部buffe的r内存大小。如果record发送的速度快于传输到server的速度，buffer将被耗尽。当buffer耗尽时，发送请求将被block。block的时长通过max.block.ms指定。在这个时间长度后，将会抛出TimeoutException

key.serialize和value.serializer用来指出如何转化key和value的值为byte。你可以使用内置的org.apache.kafka.common.serialization.ByteArraySerializer或者org.apache.kafka.common.serialization.StringSerializer处理简单的string或者byte类型

从kafka0.11开始，kafkaProducer支持两种新模式：幂等的producer和事务producer。幂等的producer让发送的语义从至少一次加强为仅一次。在特殊的producer重试时，不再会产生重复。事务性producer允许应用原子性发送消息给多个分区（及topic！）。

下一节《RecordAccumulator类代码分析》

目录：

《Kafka Producer设计分析》

《KafkaProducer类代码分析》

《RecordAccumulator类代码分析》

《Sender类代码分析》

《NetworkClient类代码分析》

-------------------------------------------------------------------

上一节《NetworkClient类代码分析》

前文我们分析了Kafka生产者端的源代码，了解了生产者产生消息的过程。消息由生产者发布到某个主题的某个分区上，其实最终是被存储在服务端的某个broker上。而消费者由订阅行为来决定它所要消费的主题和分区。消费者通过poll操作，不断的从服务端拉取该主题分区上产生的消息。

相信有兴趣看kafka源代码的同学，肯定对kafka的基本概念和原理有所了解。关于消费者，我们知道在服务端会有GroupCoordinator，负责每个consumer group的leader的选举，以及分发分区分配结果，而coumer的leader则会根据分区分配策略进行分区分配。这里需要注意，分区分配结果并不是由leader分发给同组的consumer，而是leader返回给GroupCoordinator，再有GroupCoordinator进行分发。

每当Broker有变化，或者Consumer Group有出入组的变化时，会触发ConsumerGroup的rebalance。也就是上述的分区分配工作。

另外消费者本地保存了它所负责主题分区的消费状态，通过手动和自动的方式提交到服务端的内部主题中。rebalance过后，消费者重新从内部主题获取对应主题分区的消费位置。

关于消费者更多的内容介绍请参考我另外一篇文章《Apache Kafka核心组件和流程-协调器（消费者和组协调器）》

上面我们回顾了Consumer的设计和流程，为我们进入源代码分析做好铺垫。接下来我们将从KafkaConsumer入手，进行代码分析。

KafkaConsumer

我们先看下使用KafkaConsumer进行消费的部分代码：

private final KafkaConsumer<Integer, String> consumer;

.........

consumer.subscribe(Collections.singletonList(this.topic));

ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));

for (ConsumerRecord<Integer, String> record : records) {

System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());

}

以上代码来自于源代码包中的例子，我们可以看到KafkaConsumer先订阅topic，然后通过poll方法进行消息拉取。

可以看到KafkaConsumer通过poll方法进行消费，这也是KafkaConsumer最主要的方法。

我们先看看KafkaConsumer内部的其他组件有哪些，见下图：



上图介绍了KafkaConsumer内部的几个重要组件：

1、前文说过消费者要自己记录消费的位置（但也需要提交到服务端保存，为了rebalance后的消费能衔接上），所以我们需要SubScriptionState来保存消费的状态。

2、ConsumerCoordinator负责和GroupCoordinator通讯，例如在leader选举，入组，分区分配等过程。

3、ConsumerNetworkClient是对NetworkClient的封装，如果你是从producer看过来的，想必对NetworkClient十分了解，他对nio的组件进行封装，实现网络IO。

4、PartitionAssignor，这是分区分配策略，在进行分区分配的时候会用到。

5、Fetcher负责组织拉取消息的请求，以及处理返回。不过需要注意它并不做网络IO，网络IO还是由ConsumerNetworkClient完成。它其实对应生产者中的Sender。

我们抛开订阅、rebalance这些流程，先以kafka消费流程为主，进行分析。有些组件在消费流程中是涉及不到的。消费流程主要涉及到Fetcher、SubScriptionState和ConsumerNetworkClient。特别是Fetcher，承担了重要的工作。不过我们还需要一步步来，先进入poll方法的分析。

poll()方法

这是消息拉取的入口方法，他会从上次消费的位置拉取消息，也可以手动指定消费位置。入参是阻塞的时长，如果有消息将会立即返回，否则会阻塞到超时，如果没有数据则返回空的数据集合。

代码如下：

private ConsumerRecords<K, V> poll(final Timer timer, final boolean includeMetadataInTimeout) {
    acquireAndEnsureOpen();
    try {
        if (this.subscriptions.hasNoSubscriptionOrUserAssignment()) {
            throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
        }

        // poll for new data until the timeout expires
        do {
            client.maybeTriggerWakeup();

            if (includeMetadataInTimeout) {
                if (!updateAssignmentMetadataIfNeeded(timer)) {
                    return ConsumerRecords.empty();
                }
            } else {
                while (!updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE))) {
                    log.warn("Still waiting for metadata");
                }
            }

            final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollForFetches(timer);
            if (!records.isEmpty()) {
                // before returning the fetched records, we can send off the next round of fetches
                // and avoid block waiting for their responses to enable pipelining while the user
                // is handling the fetched records.
                //
                // NOTE: since the consumed position has already been updated, we must not allow
                // wakeups or any other errors to be triggered prior to returning the fetched records.
                if (fetcher.sendFetches() > 0 || client.hasPendingRequests()) {
                    client.pollNoWakeup();
                }

                return this.interceptors.onConsume(new ConsumerRecords<>(records));
            }
        } while (timer.notExpired());

        return ConsumerRecords.empty();
    } finally {
        release();
    }
}

逻辑说明：

1、通过acquireAndEnsureOpen()确保本对象是单线程进入，这是因为KafkaConsumer非线程安全。

2、检查是否订阅了topic

3、进入主循环，条件是没有超时

4、在主循环中通过pollForFetches()拉取一次消息。这个方法中先检查是否经存在拉取过的未加工消息，这是因为上一轮次拉取做了提前拉取处理。有可能已经拉取回消息等待处理。如果没有已拉取未加工数据，则准备新的拉取请求，网络IO拉取消息，加工拉取回来的数据。

5、如果上一步拉取到消息，并不会立即返回，而是再一次触发消息拉取，并且使用的是非阻塞方式，调用client.pollNoWakeup()。这样做的目的是，提前网络IO，把消息拉取请求发出去。在网络IO的同时，消息数据返回给consumer的调用者进行业务处理。这样做到了并行处理，提高了效率。等下次调用KafkaConsumer进行poll，当进行到第4步时，有可能直接返回了上轮次提前拉取到的消息，从而省去了网络IO时间。

我们通过下图帮助理解上面4、5步的设计：



图中带颜色的方框代表在整个拉取消息的流程中，不同的处理过程，分布于不同的对象中。图中下半部分展示的是Kafka处理逻辑。可以看到在第一轮次调用了两次ConusmerNetworkClient进行IO处理，第二次IO的同时，调用者已经开始拿到返回的消息进行业务处理，这里实现了并行处理。进入第二轮次，我们发现kafkaConsumer可以直接取到上轮第二次IO回来的消息进行加工，加工后返回调用者，进行业务处理，同时下一轮次的消息拉取异步进行中。可以看到第二轮次的总时长已经没有了网络IO的时长，因为这部分工作在上一轮次已经异步进行完成。

如果不这样做，会怎么样呢？我们看图中上半部分，我们发现每个轮次都是一样的，网络IO都需要同步等待，从第二轮开始，整个消息拉取处理的时长明显增加了IO部分，会更长。

以上情况比较极端，每次提前IO都会返回数据，并且消息的业务处理时长大于网络IO。这种情况下，能最大发挥出异步IO的优势。

以上这种设计的小细节真的值得我们来学习。读源代码在了解原理的同时，我们也要多总结优秀的设计思想，对我们的工作很有帮助。

从上面的分析看到，真正消息拉取的代码是：

final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollForFetches(timer);

下面我们继续分析pollForFetches方法。

pollForFetches()方法

这个方法完成了从服务端拉取消息的动作，这个过程主要使用了Fetcher和ConsumerNetworkClient两个组件。Fetcher负责准备好拉取消息的request、处理response，并且把消息转化为对调用者友好的格式。ConsumerNetworkClient负责把请求发送出去，接收返回，也就是网络IO工作。

他的主要流程是如下四步：

1、查看是否已经存在拉取回来未加工的消息原始数据，有的话立即调用fetcher.fetchedRecords()加工，然后返回。

2、如果没有未加工的原始数据，那么调用fetcher.sendFetches()准备拉取请求。

3、通过ConsumerNetworkClient发送拉取请求。

4、加工拉取回的原始数据，返回。

其实正常来说2，3，4步流程就足够了。为什么会有第1步呢？那些已经存在的未加工的数据哪里来的？如果你理解了前面所讲的异步拉取设计，那么你应该知道答案。这些已经存在的未加工数据来自于上一轮次的异步IO。正是因为有了异步的IO拉取，才会有第一步的处理可能。

完整代码如下：

private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(Timer timer) {
    long pollTimeout = Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()), timer.remainingMs());

    // if data is available already, return it immediately
    final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchedRecords();
    if (!records.isEmpty()) {
        return records;
    }

    // send any new fetches (won't resend pending fetches)
    fetcher.sendFetches();

    // We do not want to be stuck blocking in poll if we are missing some positions
    // since the offset lookup may be backing off after a failure

    // NOTE: the use of cachedSubscriptionHashAllFetchPositions means we MUST call
    // updateAssignmentMetadataIfNeeded before this method.
    if (!cachedSubscriptionHashAllFetchPositions && pollTimeout > retryBackoffMs) {
        pollTimeout = retryBackoffMs;
    }

    Timer pollTimer = time.timer(pollTimeout);
    client.poll(pollTimer, () -> {
        // since a fetch might be completed by the background thread, we need this poll condition
        // to ensure that we do not block unnecessarily in poll()
        return !fetcher.hasCompletedFetches();
    });
    timer.update(pollTimer.currentTimeMs());

    // after the long poll, we should check whether the group needs to rebalance
    // prior to returning data so that the group can stabilize faster
    if (coordinator.rejoinNeededOrPending()) {
        return Collections.emptyMap();
    }

    return fetcher.fetchedRecords();
}

可以看到以上过程除了IO操作外，都是通过Fetcher完成的，足以体现他的重要。接下来的章节将会重点分析Fetcher。

小结

本篇先是回顾了Kafka消费者的设计，然后从KafkaConsumer的Poll方法入手对拉取的逻辑进行分析。Kafka很巧妙的采用异步IO方式，缩短整个流程的时长。接下来我们将会进入Fetcher的分析，看其如何准备拉取消息的请求，并完成消息的转化处理。

linphone版本号：3.11.1


最近在做linphone移植到hi3516d的工作，花了些时间弄懂了linphone和media2stream的运行过程，在这里分享出来，希望可以和大家一起探讨。


1.代码架构



2.代码执行流程
linphonec分析




mediastream2代码分析
1.代码结构分析

msticker分析
     概念：MSTicker是基于pthread线程封装的任务，mediastream2内部调度的对象即是MSTicker，MSTicker可以同时有多个并行运行