改名
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -36,10 +36,7 @@ class Constant
|
||||
const CONFIG_RECONNECT_BACKOFF_MAX_MS = 'reconnect.backoff.max.ms';
|
||||
const CONFIG_STATISTICS_INTERVAL_MS = 'statistics.interval.ms';
|
||||
const CONFIG_ENABLED_EVENTS = 'enabled_events';
|
||||
const CONFIG_ERROR_CB = 'error_cb';
|
||||
const CONFIG_THROTTLE_CB = 'throttle_cb';
|
||||
const CONFIG_STATS_CB = 'stats_cb';
|
||||
const CONFIG_LOG_CB = 'log_cb';
|
||||
const CONFIG_LOG_LEVEL = 'log_level';
|
||||
const CONFIG_LOG_QUEUE = 'log.queue';
|
||||
const CONFIG_LOG_THREAD_NAME = 'log.thread.name';
|
||||
@@ -115,9 +112,6 @@ class Constant
|
||||
const CONFIG_FETCH_ERROR_BACKOFF_MS = 'fetch.error.backoff.ms';
|
||||
const CONFIG_OFFSET_STORE_METHOD = 'offset.store.method';
|
||||
const CONFIG_ISOLATION_LEVEL = 'isolation.level';
|
||||
const CONFIG_CONSUME_CB = 'consume_cb';
|
||||
const CONFIG_REBALANCE_CB = 'rebalance_cb';
|
||||
const CONFIG_OFFSET_COMMIT_CB = 'offset_commit_cb';
|
||||
const CONFIG_ENABLE_PARTITION_EOF = 'enable.partition.eof';
|
||||
const CONFIG_CHECK_CRCS = 'check.crcs';
|
||||
const CONFIG_ALLOW_AUTO_CREATE_TOPICS = 'allow.auto.create.topics';
|
||||
@@ -140,7 +134,6 @@ class Constant
|
||||
const CONFIG_BATCH_SIZE = 'batch.size';
|
||||
const CONFIG_DELIVERY_REPORT_ONLY_ERROR = 'delivery.report.only.error';
|
||||
const CONFIG_DR_CB = 'dr_cb';
|
||||
const CONFIG_DR_MSG_CB = 'dr_msg_cb';
|
||||
const CONFIG_STICKY_PARTITIONING_LINGER_MS = 'sticky.partitioning.linger.ms';
|
||||
|
||||
|
||||
|
||||
+11
-15
@@ -7,12 +7,10 @@ namespace Kafka;
|
||||
use Kiri\Abstracts\Config;
|
||||
use Kiri\Exception\ConfigException;
|
||||
use Kiri\Kiri;
|
||||
use RdKafka\Conf;
|
||||
use RdKafka\Consumer;
|
||||
use RdKafka\ConsumerTopic;
|
||||
use RdKafka\Exception;
|
||||
use RdKafka\KafkaConsumer;
|
||||
use RdKafka\TopicConf;
|
||||
use Server\Abstracts\CustomProcess;
|
||||
use Swoole\Process;
|
||||
use Throwable;
|
||||
@@ -145,28 +143,26 @@ class Kafka extends CustomProcess
|
||||
private function kafkaConfig($kafka): array
|
||||
{
|
||||
try {
|
||||
$conf = new Conf();
|
||||
$conf = new Configuration();
|
||||
$conf->setRebalanceCb([$this, 'rebalanced_cb']);
|
||||
$conf->set('group.id', $kafka['groupId']);
|
||||
$conf->set('metadata.broker.list', $kafka['brokers']);
|
||||
$conf->set('socket.timeout.ms', '30000');
|
||||
|
||||
info('kafka listen groupId ' . $kafka['groupId'] . ', brokers ' . $kafka['brokers']);
|
||||
$conf->setGroupId($kafka['groupId']);
|
||||
$conf->setMetadataBrokerList($kafka['brokers']);
|
||||
$conf->setSocketTimeoutMs(30000);
|
||||
|
||||
if (function_exists('pcntl_sigprocmask')) {
|
||||
pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
|
||||
$conf->set('internal.termination.signal', (string)SIGIO);
|
||||
$conf->setInternalTerminationSignal((string)SIGIO);
|
||||
}
|
||||
|
||||
$topicConf = new TopicConf();
|
||||
$topicConf->set('auto.commit.enable', '1');
|
||||
$topicConf->set('auto.commit.interval.ms', '100');
|
||||
$topicConf = new TopicConfig();
|
||||
$topicConf->setAutoCommitEnable(true);
|
||||
$topicConf->setAutoCommitIntervalMs(100);
|
||||
|
||||
//smallest：简单理解为从头开始消费，
|
||||
//largest：简单理解为从最新的开始消费
|
||||
$topicConf->set('auto.offset.reset', 'smallest');
|
||||
$topicConf->set('offset.store.path', 'kafka_offset.log');
|
||||
$topicConf->set('offset.store.method', 'broker');
|
||||
$topicConf->setAutoOffsetReset('smallest');
|
||||
$topicConf->setOffsetStorePath('kafka_offset.log');
|
||||
$topicConf->setOffsetStoreMethod('broker');
|
||||
|
||||
return [$conf, $topicConf, $kafka];
|
||||
} catch (Throwable $exception) {
|
||||
|
||||
+33
-16
@@ -3,15 +3,13 @@
|
||||
namespace Kafka;
|
||||
|
||||
use Exception;
|
||||
use RdKafka\Conf;
|
||||
use RdKafka\Producer;
|
||||
use RdKafka\ProducerTopic;
|
||||
use RdKafka\TopicConf;
|
||||
use ReflectionException;
|
||||
use Kiri\Abstracts\Config;
|
||||
use Kiri\Exception\ConfigException;
|
||||
use Kiri\Exception\NotFindClassException;
|
||||
use Kiri\Kiri;
|
||||
use RdKafka\Producer;
|
||||
use RdKafka\ProducerTopic;
|
||||
use ReflectionException;
|
||||
|
||||
|
||||
/**
|
||||
@@ -21,8 +19,8 @@ class KafkaClient
|
||||
{
|
||||
|
||||
|
||||
private Conf $conf;
|
||||
private TopicConf $topicConf;
|
||||
private Configuration $conf;
|
||||
private TopicConfig $topicConf;
|
||||
|
||||
private bool $isAck = true;
|
||||
|
||||
@@ -36,24 +34,43 @@ class KafkaClient
|
||||
*/
|
||||
public function __construct(public string $topic, public string $groupId)
|
||||
{
|
||||
$this->conf = di(Conf::class);
|
||||
$this->topicConf = di(TopicConf::class);
|
||||
$this->setConfig($this->conf);
|
||||
$this->conf = di(Configuration::class);
|
||||
$this->topicConf = di(TopicConfig::class);
|
||||
$this->setConfig();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @return TopicConfig
|
||||
*/
|
||||
public function getTopicConfig(): TopicConfig
|
||||
{
|
||||
return $this->topicConf;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @return Configuration
|
||||
*/
|
||||
public function getConfiguration(): Configuration
|
||||
{
|
||||
return $this->conf;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @throws ConfigException
|
||||
*/
|
||||
private function setConfig(Conf $kafkaConfig)
|
||||
private function setConfig()
|
||||
{
|
||||
$config = Config::get('kafka.producers.' . $this->topic, null, true);
|
||||
if (!isset($config['brokers']) || !isset($config['groupId'])) {
|
||||
if (!isset($config['brokers'])) {
|
||||
throw new ConfigException('Please configure relevant information.');
|
||||
}
|
||||
$kafkaConfig->set('metadata.broker.list', $config['brokers']);
|
||||
$kafkaConfig->set('group.id', $this->groupId);
|
||||
$kafkaConfig->setErrorCb(function ($kafka, $err, $reason) {
|
||||
$this->conf->setMetadataBrokerList($config['brokers']);
|
||||
$this->conf->setGroupId($this->groupId);
|
||||
$this->conf->setClientId(current(swoole_get_local_ip()));
|
||||
$this->conf->setErrorCb(function ($kafka, $err, $reason) {
|
||||
logger()->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
|
||||
});
|
||||
}
|
||||
@@ -101,7 +118,7 @@ class KafkaClient
|
||||
*/
|
||||
private function getProducerTopic(Producer $producer, $topic, $isAck): ProducerTopic
|
||||
{
|
||||
$this->topicConf->set('request.required.acks', $isAck ? '1' : '0');
|
||||
$this->topicConf->setRequestRequiredAcks($isAck ? '1' : '0');
|
||||
return $producer->newTopic($topic, $this->topicConf);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,287 @@
|
||||
<?php
|
||||
|
||||
namespace Kafka;
|
||||
|
||||
use RdKafka\TopicConf;
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
class TopicConfig extends TopicConf
|
||||
{
|
||||
|
||||
/**
|
||||
* @param mixed|string $request_required_acks
|
||||
*
|
||||
* 此配置是表明当一次produce请求被认为完成时的确认值。特别是，多少个其他brokers必须已经提交了数据到他们的log并且向他们的leader确认了这些信息。典型的值包括：
|
||||
* 0： 表示producer从来不等待来自broker的确认信息（和0.7一样的行为）。这个选择提供了最小的时延但同时风险最大（因为当server宕机时，数据将会丢失）。
|
||||
* 1：表示获得leader replica已经接收了数据的确认信息。这个选择时延较小同时确保了server确认接收成功。
|
||||
* -1：producer会获得所有同步replicas都收到数据的确认。同时时延最大，然而，这种方式并没有完全消除丢失消息的风险，因为同步replicas的数量可能是1.如果你想确保某些replicas接收到数据，那么你应该在topic-level设置中选项min.insync.replicas设置一下。请阅读一下设计文档，可以获得更深入的讨论。
|
||||
*/
|
||||
public function setRequestRequiredAcks(int $request_required_acks): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_REQUEST_REQUIRED_ACKS, $request_required_acks);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @param mixed|string $acks
|
||||
*
|
||||
* producer需要server接收到数据之后发出的确认接收的信号，此项配置就是指procuder需要多少个这样的确认信号。此配置实际上代表了数据备份的可用性。以下设置为常用选项：
|
||||
* （1）acks=0： 设置为0表示producer不需要等待任何确认收到的信息。副本将立即加到socket buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回馈的offset会总是设置为-1；
|
||||
* （2）acks=1： 这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
|
||||
* （3）acks=all： 这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
|
||||
* （4）其他的设置，例如acks=2也是可以的，这将需要给定的acks数量，但是这种策略一般很少用。
|
||||
*/
|
||||
public function setAcks(int $acks): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_ACKS, $acks);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $request_timeout_ms
|
||||
*
|
||||
* broker尽力实现request.required.acks需求时的等待时间，否则会发送错误到客户端
|
||||
*/
|
||||
public function setRequestTimeoutMs(int $request_timeout_ms): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_REQUEST_TIMEOUT_MS, $request_timeout_ms);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $message_timeout_ms
|
||||
*
|
||||
* 本地消息超时。此值仅在本地强制执行，并限制生成的消息等待成功传递的时间。0的时间是无限的。
|
||||
* 这是librdkafka用于传递消息（包括重试）的最长时间。
|
||||
* 超过重试计数或消息超时时发生传递错误。
|
||||
* 如果配置了transactional.id，则消息超时将自动调整为transaction.timeout.ms。
|
||||
*/
|
||||
public function setMessageTimeoutMs(int $message_timeout_ms): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_MESSAGE_TIMEOUT_MS, $message_timeout_ms);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $delivery_timeout_ms
|
||||
*
|
||||
* Alias for message.timeout.ms: Local message timeout.
|
||||
* This value is only enforced locally and limits the time a produced message waits for successful delivery.
|
||||
* A time of 0 is infinite.
|
||||
* This is the maximum time librdkafka may use to deliver a message (including retries).
|
||||
* Delivery error occurs when either the retry count or the message timeout are exceeded.
|
||||
* The message timeout is automatically adjusted to transaction.timeout.ms if transactional.id is configured.
|
||||
*/
|
||||
public function setDeliveryTimeoutMs(int $delivery_timeout_ms): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_DELIVERY_TIMEOUT_MS, $delivery_timeout_ms);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $queuing_strategy
|
||||
*
|
||||
* EXPERIMENTAL: subject to change or removal.
|
||||
* DEPRECATED Producer queuing strategy.
|
||||
* FIFO preserves produce ordering, while LIFO prioritizes new messages
|
||||
*/
|
||||
public function setQueuingStrategy(mixed $queuing_strategy): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_QUEUING_STRATEGY, $queuing_strategy);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $produce_offset_report
|
||||
*
|
||||
* DEPRECATED No longer used.
|
||||
*/
|
||||
public function setProduceOffsetReport(bool $produce_offset_report): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_PRODUCE_OFFSET_REPORT, $produce_offset_report);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $partitioner
|
||||
*
|
||||
* Partitioner: random - random distribution,
|
||||
* consistent - CRC32 hash of key (Empty and NULL keys are mapped to single partition),
|
||||
* consistent_random - CRC32 hash of key (Empty and NULL keys are randomly partitioned),
|
||||
* murmur2 - Java Producer compatible Murmur2 hash of key (NULL keys are mapped to single partition),
|
||||
* murmur2_random - Java Producer compatible Murmur2 hash of key (NULL keys are randomly partitioned.
|
||||
* This is functionally equivalent to the default partitioner in the Java Producer.),
|
||||
* fnv1a - FNV-1a hash of key (NULL keys are mapped to single partition),
|
||||
* fnv1a_random - FNV-1a hash of key (NULL keys are randomly partitioned).
|
||||
*/
|
||||
public function setPartitioner(mixed $partitioner): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_PARTITIONER, $partitioner);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $partitioner_cb
|
||||
*
|
||||
* Custom partitioner callback (set with rd_kafka_topic_conf_set_partitioner_cb())
|
||||
*/
|
||||
public function setPartitionerCb(mixed $partitioner_cb): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_PARTITIONER_CB, $partitioner_cb);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $msg_order_cmp
|
||||
*
|
||||
* EXPERIMENTAL: subject to change or removal.
|
||||
* DEPRECATED Message queue ordering comparator (set with rd_kafka_topic_conf_set_msg_order_cmp()).
|
||||
* Also see queuing.strategy.
|
||||
*
|
||||
*/
|
||||
public function setMsgOrderCmp(mixed $msg_order_cmp): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_MSG_ORDER_CMP, $msg_order_cmp);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $opaque
|
||||
*
|
||||
* Application opaque (set with rd_kafka_topic_conf_set_opaque())
|
||||
*/
|
||||
public function setOpaque(mixed $opaque): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_OPAQUE, $opaque);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $compression_codec
|
||||
*
|
||||
* Compression codec to use for compressing message sets. inherit = inherit global compression.codec configuration.
|
||||
*/
|
||||
public function setCompressionCodec(mixed $compression_codec): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_COMPRESSION_CODEC, $compression_codec);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $compression_type
|
||||
*
|
||||
* Alias for compression.codec: compression codec to use for compressing message sets.
|
||||
* This is the default value for all topics, may be overridden by the topic configuration property compression.codec.
|
||||
*/
|
||||
public function setCompressionType(mixed $compression_type): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_COMPRESSION_TYPE, $compression_type);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $compression_level
|
||||
*
|
||||
* Compression level parameter for algorithm selected by configuration property compression.codec.
|
||||
* Higher values will result in better compression at the cost of more CPU usage.
|
||||
* Usable range is algorithm-dependent: [0-9] for gzip; [0-12] for lz4; only 0 for snappy;
|
||||
* -1 = codec-dependent default compression level.
|
||||
*/
|
||||
public function setCompressionLevel(int $compression_level): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_COMPRESSION_LEVEL, $compression_level);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $auto_commit_enable
|
||||
*
|
||||
* DEPRECATED [LEGACY PROPERTY: This property is used by the simple legacy consumer only.
|
||||
* When using the high-level KafkaConsumer, the global enable.auto.commit property must be used instead].
|
||||
* If true, periodically commit offset of the last message handed to the application.
|
||||
* This committed offset will be used when the process restarts to pick up where it left off.
|
||||
* If false, the application will have to call rd_kafka_offset_store() to store an offset (optional).
|
||||
* Offsets will be written to broker or local file according to offset.store.method.
|
||||
*/
|
||||
public function setAutoCommitEnable(bool $auto_commit_enable): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_AUTO_COMMIT_ENABLE, $auto_commit_enable);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $enable_auto_commit
|
||||
*
|
||||
* DEPRECATED Alias for auto.commit.enable: [LEGACY PROPERTY: This property is used by the simple legacy consumer only.
|
||||
* When using the high-level KafkaConsumer, the global enable.auto.commit property must be used instead].
|
||||
* If true, periodically commit offset of the last message handed to the application.
|
||||
* This committed offset will be used when the process restarts to pick up where it left off.
|
||||
* If false, the application will have to call rd_kafka_offset_store() to store an offset (optional).
|
||||
* Offsets will be written to broker or local file according to offset.store.method.
|
||||
*/
|
||||
public function setEnableAutoCommit(bool $enable_auto_commit): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_ENABLE_AUTO_COMMIT, $enable_auto_commit);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $auto_commit_interval_ms
|
||||
*
|
||||
* [LEGACY PROPERTY: This setting is used by the simple legacy consumer only.
|
||||
* When using the high-level KafkaConsumer, the global auto.commit.interval.ms property must be used instead].
|
||||
* The frequency in milliseconds that the consumer offsets are committed (written) to offset storage.
|
||||
*/
|
||||
public function setAutoCommitIntervalMs(int $auto_commit_interval_ms): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_AUTO_COMMIT_INTERVAL_MS, $auto_commit_interval_ms);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $auto_offset_reset
|
||||
*
|
||||
* Action to take when there is no initial offset in offset store or the desired offset is out of range:
|
||||
* 'smallest','earliest' - automatically reset the offset to the smallest offset,
|
||||
* 'largest','latest' - automatically reset the offset to the largest offset,
|
||||
* 'error' - trigger an error (ERR__AUTO_OFFSET_RESET) which is retrieved by consuming messages and checking 'message->err'.
|
||||
*/
|
||||
public function setAutoOffsetReset(mixed $auto_offset_reset): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_AUTO_OFFSET_RESET, $auto_offset_reset);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $offset_store_path
|
||||
*
|
||||
* DEPRECATED Path to local file for storing offsets.
|
||||
* If the path is a directory a filename will be automatically generated in that directory based on the topic and partition.
|
||||
* File-based offset storage will be removed in a future version.
|
||||
*/
|
||||
public function setOffsetStorePath(string $offset_store_path): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_OFFSET_STORE_PATH, $offset_store_path);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $offset_store_sync_interval_ms
|
||||
*
|
||||
* DEPRECATED fsync() interval for the offset file, in milliseconds.
|
||||
* Use -1 to disable syncing, and 0 for immediate sync after each write.
|
||||
* File-based offset storage will be removed in a future version.
|
||||
*/
|
||||
public function setOffsetStoreSyncIntervalMs(int $offset_store_sync_interval_ms): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_OFFSET_STORE_SYNC_INTERVAL_MS, $offset_store_sync_interval_ms);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $offset_store_method
|
||||
*
|
||||
* DEPRECATED Offset commit store method:
|
||||
* 'file' - DEPRECATED: local file store (offset.store.path, et.al),
|
||||
* 'broker' - broker commit store (requires "group.id" to be configured and Apache Kafka 0.8.2 or later on the broker.).
|
||||
*/
|
||||
public function setOffsetStoreMethod(mixed $offset_store_method): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_OFFSET_STORE_METHOD, $offset_store_method);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mixed|string $consume_callback_max_messages
|
||||
*
|
||||
* Maximum number of messages to dispatch in one rd_kafka_consume_callback*()
|
||||
*/
|
||||
public function setConsumeCallbackMaxMessages(int $consume_callback_max_messages): void
|
||||
{
|
||||
$this->set(Constant::TOPIC_CONF_CONSUME_CALLBACK_MAX_MESSAGES, $consume_callback_max_messages);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
<?php
|
||||
|
||||
|
||||
return [
|
||||
'kafka' => [
|
||||
|
||||
''
|
||||
|
||||
|
||||
]
|
||||
];
|
||||
Reference in New Issue
Block a user