1. 直接队列

直接队列是非排队队列。直接模式下的队列既不排队也不缓冲任何队列元素，而是直接（并立即）将元素从生产者传递给消费者。这听起来很奇怪，但这种队列类型有充分的理由。

直接模式队列允许一般使用队列，即使在不总是需要排队的地方也是如此。一个很好的例子是输出动作前面的队列。虽然缓冲转发操作或数据库写入非常有意义，但在简单的本地文件写入之前构建队列的意义有限。然而，rsyslog仍然在每个动作前都有一个队列。因此，对于文件写入，队列模式可以简单地设置为“直接”，在这种情况下不会发生排队。

请注意，直接队列也是唯一一种将执行返回代码（成功/失败）从消费者传递回生产者的队列类型。例如，备份操作逻辑需要这样做。因此，备份操作要求待检查操作使用“直接”模式队列。

2. 磁盘队列

磁盘队列使用磁盘驱动器进行缓冲。重要的事实是它们总是使用磁盘并且不在内存中缓冲任何内容。因此，队列是超可靠的，但到目前为止是最慢的模式。对于常规用例，不建议使用此队列模式。如果日志数据非常重要，即使在极端情况下也不会丢失，这很有用。

写入磁盘队列时，它以块的形式完成。每个块都接收其单个文件。文件以前缀命名（通过“ $ <object> QueueFilename ”配置指令设置），后跟一个7位数字（从1开始，每个文件递增）。块默认为10mb，可以通过“ $ <object> QueueMaxFileSize ”配置指令设置不同的大小。请注意，大小限制不是很明显：rsyslog总是写入一个完整的队列条目，即使它违反了大小限制。所以块实际上比配置的大小稍微大一点（通常小于1k）。出于同样的原因，每个块也具有不同的大小。如果您观察到不同的块大小，您可以放松：这不是问题。

使用以块为单位进行写入，以便可以快速删除已处理的数据并且可以免费用于其他用途 - 同时保持对所使用的磁盘空间没有人为的上限。如果设置了磁盘配额（下面的说明），请确保配额/块大小允许写入至少两个块。Rsyslog目前没有检查，如果一个块超过配额，将会失败。

创建新块会降低性能，但可以更快地释放磁盘空间。10mb的默认值被认为是这两者之间的良好折衷。但是，将这些设置适应本地策略可能是有意义的。例如，如果磁盘队列写在专用的200gb磁盘上，则使用2gb（甚至更大）的块大小可能是有意义的。

但请注意，默认情况下磁盘队列在每次写入磁盘时都不会更新其内务处理结构。这是出于性能原因。如果发生故障，数据仍将丢失（除非手动修复文件结构）。但是，可以将磁盘队列设置为在检查点（每n个记录）上写入簿记信息，这样也可以使其更加可靠。如果检查点间隔设置为1，则不会丢失任何数据，但队列异常缓慢。

每个队列都可以放在不同的磁盘上，以获得最佳性能和/或隔离。目前通过在队列创建语句之前指定不同的$ WorkDirectory配置指令来选择此选项 。

要创建磁盘队列，请使用“ $ <object> QueueType Disk ”配置指令。可以通过“ $ <object> QueueCheckpointInterval ” 指定检查点间隔，0表示没有检查点。请注意，通过在每次写入操作后发出（f）同步，可以使基于磁盘的队列非常可靠。从版本4.3.2开始，可以通过“ <object> QueueSyncQueueFiles on / off，默认为关闭”来请求。激活此选项会降低性能，因此不应无故启用。

3. 内存中队列

内存队列模式是大多数人在考虑计算队列时的想法。这里，排队的数据元素保存在存储器中。因此，内存中的队列非常快。但是，当然，它们无法在任何程序或操作系统中止（通常是可以容忍的并且不太可能）。如果使用内存模式，请确保使用UPS，并且日志数据对您很重要。请注意，即使内存中的队列可能会在无限量的时间内保存数据，例如输出目标系统已关闭且没有理由将数据移出内存（在内存中长时间处于非存储状态）原因）。纯内存队列甚至无法将队列元素存储在核心内存中的任何其他位置。

存在两种不同的内存中队列模式：LinkedList和FixedArray。从用户的角度来看，两者都非常相似，但使用不同的算法。

FixedArray队列使用固定的预先分配的数组，该数组保存指向队列元素的指针。大多数空间由实际用户数据元素占用，数组中的指针指向该元素。指针数组本身相对较小。但是，即使队列为空，它也有一定的内存占用。由于无需动态分配任何内务处理结构，因此FixedArray提供了最佳的运行时性能（使用最少的CPU周期）。如果期望的队列元素数量相对较少且需要性能，则FixedArray是最佳的。它是主消息队列的默认模式（限制为10,000个元素）。

LinkedList队列恰恰相反。所有管家结构都是动态分配的（在链接列表中，顾名思义）。这需要更多的运行时处理开销，但确保仅在需要时分配内存。LinkedList队列尤其适用于只需偶尔需要排队大量元素的队列。用例可能是偶尔的消息突发。在内存允许的情况下，它可以限制为例如200,000个元素，如果在使用中则仅占用内存。在这种情况下，FixedArray队列可能具有太大的静态内存占用。

通常，如果有疑问，建议使用LinkedList模式。与FixedArray相比，处理开销很低，并且可能因内存使用的减少而被抵消。在大多数经常未使用的指针数组页面中进行分页可能比动态分配它们要慢得多。

要创建内存中队列，请使用“ $ <object> QueueType LinkedList ”或“ $ <object> QueueType FixedArray ”配置指令。

<?php

/**

 * Created by IntelliJ IDEA.

 * User: qiangjian

 * Date: 2019/3/14

 * Time: 09:59

 */

namespace Lib\Logger\Adapter;

class Syslog implements IWriter {

    protected $name;

    protected static $instances;

    private function __construct($name)

    {

        $this->name = $name;

    }

    private function __clone(){}

    static function create($name){

        if(isset(self::$instances[$name])){

            return self::$instances[$name];

        }

        return self::$instances[$name] = new self($name);

    }

    function write($msgStr, $level = LOG_INFO){

        openlog($this->name,  LOG_ODELAY, LOG_USER);

        return syslog($level, $msgStr);

    }

}

module(load="imuxsock") # provides support for local system logging (e.g. via logger command)

module(load="imklog")   # provides kernel logging support (previously done by rklogd)

module(load="impstats"

    interval=""

    severity=""

    resetCounters="on"

    log.syslog="off"

    log.file="/data1/ms/log/rsyslog/rsyslog_status.log")

*.info;mail.none;authpriv.none;cron.none;local6.none;local5.none;user.none                 /var/log/messages

# nginx template

template(name="kfk_log" type="string" string="%msg:2:$%")

# ruleset

ruleset(name="to_bip_kfk") {

    #日志转发kafka

    action(

        name="to_bip_kfk"

        type="omkafka"

        topic="cms-front-network"

        action.resumeRetryCount="-1"

        action.ResumeInterval=""

        broker=["xxxxxx:9110"]

    confParam=["sasl.mechanisms=PLAIN","security.protocol=sasl_plaintext","sasl.username=xxx","sasl.password=xxx","debug=all","api.version.request=true"]

        partitions.auto="on"

        template="kfk_log"

        errorFile="/data1/ms/log/rsyslog/to_kfk_err.log"

        queue.filename="queue_filename"

        queue.type="linkedlist"

        queue.spoolDirectory="/data1/ms/log/rsyslog" #提前建目录

        queue.workerthreads=""

        queue.workerthreadminimummessages=""

        queue.size=""

        queue.highwatermark=""

        queue.LowWatermark="" #当内存队列小于这些元素时，停止回写磁盘。

        queue.TimeoutEnqueue=""

        queue.dequeuebatchsize=""

        queue.saveonshutdown="on"

        )

}

if $programname == 'front_network' then {

  #/data1/debug.log

    call to_bip_kfk

        stop

}

> strace -p  -fciT

Process  attached with  threads - interrupt to quit

^CProcess  detached

Process  detached

Process  detached

Process  detached

Process  detached

% time     seconds  usecs/call     calls    errors syscall

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

 61.06   60.070903                        poll

 35.23   34.655613                    futex

  1.53    1.505787                      read

  1.29    1.265297                          select

  0.89    0.871452                        restart_syscall

  0.00    0.003077                       recvmsg

  0.00    0.001816                            write

  0.00    0.001477                            sendmsg

  0.00    0.000386                            stat

  0.00    0.000021                             madvise

  0.00    0.000000                              close

  0.00    0.000000                              socket

  0.00    0.000000                            connect

  0.00    0.000000                              sendto

  0.00    0.000000                              bind

  0.00    0.000000                              getsockname

  0.00    0.000000                              getsockopt

  0.00    0.000000                              fcntl

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

100.00   98.375829                      total

rsyslogd -dn |grep -i 'err|fail'

rsyslogd: error in kafka parameter 'sasl.mechanisms=PLAIN': No such configuration property: "sasl.mechanisms" [v8.25.0..

wget  https://github.com/edenhill/librdkafka/archive/master.zip

unzip master

cd librdkafka-master/

ls

./configure

make

make install 
ldconfig 

strings -a /usr/local/lib/librdkafka.so  |grep -E '[0-9]\.[0-9]+\.[0-9]+$' #确认下版本，不低于0.9的kafka即可

rsyslogd -dn |grep -i 'err|fail'