Skip to content
This repository was archived by the owner on Jun 10, 2022. It is now read-only.

数据丢失 & 单例模式 bug #251

Closed
wants to merge 5 commits into from
Closed

数据丢失 & 单例模式 bug #251

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
705f1dd
fix: offset + 1
noname007 Aug 1, 2018
fc3b3f1
Fix bug: broker instance will be wrong used when the consumer and pro…
noname007 Sep 14, 2018
4ae5dbb
php version restrict
noname007 Sep 14, 2018
c277bef
fix
noname007 Sep 14, 2018
385678d
Fix producer call
noname007 Sep 14, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion composer.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
"sort-packages": true
},
"require": {
"php": "^5.6",
"php": ">=5.6",
"amphp/amp": "^v1.2.2",
"psr/log": "^1.0.2"
},
Expand Down
18 changes: 16 additions & 2 deletions src/Broker.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,6 @@

class Broker
{
use SingletonTrait;

private $groupBrokerId = null;

private $topics = [];
Expand All @@ -23,6 +21,22 @@ class Broker

private $config;

private static $instance = [];

/**
* @return static
*/
public static function getInstance($instance_name = 'default')
{

if(isset(self::$instance[$instance_name] )) {
return self::$instance[$instance_name];
}else{
return self::$instance[$instance_name] = new static();
}
}


public function setProcess(callable $process)
{
$this->process = $process;
Expand Down
15 changes: 6 additions & 9 deletions src/Consumer/Process.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -490,7 +490,7 @@ public function succFetchOffset($result)
foreach ($consumerOffsets as $topic => $value) {
foreach ($value as $partId => $offset) {
if (isset($lastOffsets[$topic][$partId]) && $lastOffsets[$topic][$partId] > $offset) {
$consumerOffsets[$topic][$partId] = $offset + 1;
$consumerOffsets[$topic][$partId] = $offset;
}
}
}
Expand Down Expand Up @@ -557,23 +557,20 @@ public function succFetch($result, $fd)
continue;
}

$consumerOffset = $assign->getConsumerOffset($topic['topicName'], $part['partition']);
if ($consumerOffset === false) {
$offset = $assign->getConsumerOffset($topic['topicName'], $part['partition']);
if ($offset === false) {
return; // current is rejoin....
}
foreach ($part['messages'] as $message) {
$this->messages[$topic['topicName']][$part['partition']][] = $message;
//if ($this->consumer != null) {
// call_user_func($this->consumer, $topic['topicName'], $part['partition'], $message);
//}
$commitOffset = $message['offset'];
$offset = $message['offset'] + 1;
}

$commitOffset = isset($commitOffset) ? $commitOffset : $consumerOffset - 1;
$consumerOffset = $commitOffset + 1;

$assign->setConsumerOffset($topic['topicName'], $part['partition'], $consumerOffset);
$assign->setCommitOffset($topic['topicName'], $part['partition'], $commitOffset);
$assign->setConsumerOffset($topic['topicName'], $part['partition'], $offset);
$assign->setCommitOffset($topic['topicName'], $part['partition'], $offset);
}
}
$this->state->succRun(\Kafka\Consumer\State::REQUEST_FETCH, $fd);
Expand Down
10 changes: 5 additions & 5 deletions src/Producer/Process.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ public function init()
\Kafka\Protocol::init($config->getBrokerVersion(), $this->logger);

// init process request
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$broker->setConfig($config);
$broker->setProcess(function ($data, $fd) {
$this->processRequest($data, $fd);
Expand Down Expand Up @@ -124,7 +124,7 @@ public function syncMeta()
}

shuffle($brokerHost);
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
foreach ($brokerHost as $host) {
$socket = $broker->getMetaConnect($host);
if ($socket) {
Expand Down Expand Up @@ -160,7 +160,7 @@ protected function processRequest($data, $fd)
$this->error('Get metadata is fail, brokers or topics is null.');
$this->state->failRun(\Kafka\Producer\State::REQUEST_METADATA);
} else {
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$isChange = $broker->setData($result['topics'], $result['brokers']);
$this->state->succRun(\Kafka\Producer\State::REQUEST_METADATA, $isChange);
}
Expand All @@ -177,7 +177,7 @@ protected function processRequest($data, $fd)
protected function produce()
{
$context = [];
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$requiredAck = \Kafka\ProducerConfig::getInstance()->getRequiredAck();
$timeout = \Kafka\ProducerConfig::getInstance()->getTimeout();

Expand Down Expand Up @@ -260,7 +260,7 @@ protected function stateConvert($errorCode, $context = null)
protected function convertMessage($data)
{
$sendData = [];
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$topicInfos = $broker->getTopics();
foreach ($data as $value) {
if (! isset($value['topic']) || ! trim($value['topic'])) {
Expand Down
10 changes: 5 additions & 5 deletions src/Producer/SyncProcess.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,15 @@ public function __construct()
$config = \Kafka\ProducerConfig::getInstance();
\Kafka\Protocol::init($config->getBrokerVersion(), $this->logger);
// init broker
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$broker->setConfig($config);

$this->syncMeta();
}

public function send($data)
{
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$requiredAck = \Kafka\ProducerConfig::getInstance()->getRequiredAck();
$timeout = \Kafka\ProducerConfig::getInstance()->getTimeout();

Expand Down Expand Up @@ -80,7 +80,7 @@ public function syncMeta()
}

shuffle($brokerHost);
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
foreach ($brokerHost as $host) {
$socket = $broker->getMetaConnect($host, true);
if ($socket) {
Expand All @@ -95,7 +95,7 @@ public function syncMeta()
if (! isset($result['brokers']) || ! isset($result['topics'])) {
throw new \Kafka\Exception('Get metadata is fail, brokers or topics is null.');
} else {
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$broker->setData($result['topics'], $result['brokers']);
}
return;
Expand All @@ -113,7 +113,7 @@ public function syncMeta()
protected function convertMessage($data)
{
$sendData = [];
$broker = \Kafka\Broker::getInstance();
$broker = \Kafka\Broker::getInstance(__CLASS__);
$topicInfos = $broker->getTopics();
foreach ($data as $value) {
if (! isset($value['topic']) || ! trim($value['topic'])) {
Expand Down