see #1 #2 #3 #4

weiboad · May 31, 2017 · f5492bd · f5492bd
1 parent c498b78
commit f5492bd
Show file tree

Hide file tree

Showing 17 changed files with 457 additions and 106 deletions.
diff --git a/conf/system.ini b/conf/system.ini
@@ -9,7 +9,7 @@ logsDir=logs/
 ; 日志分卷大小
 logRollSize=52428800
 ; 1: LOG_TRACE 2: LOG_DEBUG 3: LOG_INFO 
-logLevel=1
+logLevel=3
 isAsync=no
 
 [adserver]
@@ -28,7 +28,7 @@ serverName=mc-server
 
 [http]
 host=0.0.0.0
-port=20010
+port=10053
 timeout=3
 threadNum=4
 serverName=adinf-adserver
@@ -45,14 +45,11 @@ threadNum=4
 serverName=head-server
 
 [kafkac_out]
-;是否是 v0.9.x 新协议获取
-isNewConsumerOut=yes
 ; 支持同时消费多个 topic, 多个用逗号分隔
-topicNameOut=test
+topicNameOut=fa_status_add
 groupIdOut=test_group_id
-brokerListOut=10.13.4.161:9192,10.13.4.160:9192
+brokerListOut=10.77.96.136:9192,10.77.96.137:9192
 kafkaDebugOut=none
-offsetPathOut=./
 statIntervalOut=60000
 
 [timer]

diff --git a/src/AdbaseConfig.hpp b/src/AdbaseConfig.hpp
@@ -14,9 +14,7 @@
 	std::string groupId##name;\
 	std::string brokerListConsumer##name;\
 	std::string kafkaDebug##name;\
-	std::string offsetPath##name;\
-	std::string statInterval##name;\
-	bool isNewConsumer##name;
+	std::string statInterval##name;
 #endif
 #ifndef DECLARE_KAFKA_PRODUCER_CONFIG
 #define DECLARE_KAFKA_PRODUCER_CONFIG(name) \
@@ -85,6 +83,7 @@ class App;
 namespace app {
 	class Message;
 	class Storage;
+	class Metrics;
 }
 typedef struct adserverContext {
 	AdbaseConfig* config;
@@ -93,6 +92,7 @@ typedef struct adserverContext {
 	adbase::metrics::Metrics* metrics;
 	// 前后端交互数据指针添加到下面
 	app::Storage* storage;
+	app::Metrics* appMetrics;
 } AdServerContext;
 
 typedef struct aimsContext {
@@ -101,13 +101,15 @@ typedef struct aimsContext {
 	// 消息队列交互上下文
 	app::Message* message;
 	app::Storage* storage;
+	app::Metrics* appMetrics;
 } AimsContext;
 typedef struct timerContext {
 	AdbaseConfig* config;
 	adbase::EventBasePtr mainEventBase;	
 	App* app;
 	// 定时器交互上下文
 	app::Storage* storage;
+	app::Metrics* appMetrics;
 } TimerContext;
 
 #endif
diff --git a/src/Aims.cpp b/src/Aims.cpp
@@ -1,37 +1,5 @@
 #include "Aims.hpp"
 
-// {{{ macros
-
-#define STOP_KAFKA_CONSUMER(name) do {\
-} while(0)
-
-#define INIT_KAFKA_PRODUCER(name) do {\
-	_kafkaProducerCallback##name = new aims::kafka::Producer##name(_context);\
-	_kafkaProducer##name = new adbase::kafka::Producer(_configure->brokerListProducer##name,\
-						   _configure->queueLength##name, _configure->debug##name);\
-	_kafkaProducer##name->setSendHandler(std::bind(&aims::kafka::Producer##name::send,\
-											   _kafkaProducerCallback##name,\
-											   std::placeholders::_1, std::placeholders::_2,\
-											   std::placeholders::_3, std::placeholders::_4));\
-	_kafkaProducer##name->setAckHandler(std::bind(&aims::kafka::Producer##name::ackCallback, \
-													 _kafkaProducerCallback##name, std::placeholders::_1));\
-	_kafkaProducer##name->setErrorHandler(std::bind(&aims::kafka::Producer##name::errorCallback, \
-													 _kafkaProducerCallback##name, std::placeholders::_1));\
-} while(0)
-#define START_KAFKA_PRODUCER(name) do {\
-	_kafkaProducer##name->start();\
-} while(0)
-#define STOP_KAFKA_PRODUCER(name) do {\
-	if (_kafkaProducer##name != nullptr) {\
-		_kafkaProducer##name->stop();\
-	}\
-	if (_kafkaProducerCallback##name != nullptr) {\
-		delete _kafkaProducerCallback##name;\
-		_kafkaProducerCallback##name = nullptr;\
-	}\
-} while(0)
-
-// }}}
 // {{{ Aims::Aims()
 
 Aims::Aims(AimsContext* context) :
@@ -52,10 +20,15 @@ Aims::~Aims() {
 void Aims::run() {
 	// 初始化 server
 	init();
+    _kafka->start();
+}
 
-	for(auto &t : _kafkas) {
-		t.second->start();	
-	}
+// }}}
+// {{{ void Aims::reload()
+
+void Aims::reload() {
+	std::vector<std::string> topicNames = adbase::explode(_configure->topicNameConsumerOut, ',', true);
+    _kafka->setTopics(topicNames);
 }
 
 // }}}
@@ -69,11 +42,10 @@ void Aims::init() {
 // {{{ void Aims::stop()
 
 void Aims::stop() {
-	for (auto &t : _kafkas) {
-		if (t.second != nullptr) {
-			t.second->stop();
-		}
-	}
+    if (_kafka != nullptr) {
+        _kafka->stop();
+        delete _kafka;
+    }
 	if (_kafkaConsumerCallbackOut != nullptr) {
 		delete _kafkaConsumerCallbackOut;
 		_kafkaConsumerCallbackOut = nullptr;
@@ -86,25 +58,19 @@ void Aims::stop() {
 void Aims::initKafkaConsumer() {
 	_kafkaConsumerCallbackOut = new aims::kafka::ConsumerOut(_context);
 	std::vector<std::string> topicNames = adbase::explode(_configure->topicNameConsumerOut, ',', true);
-	for(auto &t : topicNames) {
-		adbase::kafka::Consumer* consumer = new adbase::kafka::Consumer(t, _configure->groupIdOut, 
-											_configure->brokerListConsumerOut);	
-		consumer->setMessageHandler(std::bind(&aims::kafka::ConsumerOut::pull,
-												_kafkaConsumerCallbackOut,
-												std::placeholders::_1, std::placeholders::_2,
-												std::placeholders::_3, std::placeholders::_4));
-		consumer->setStatCallback(std::bind(&aims::kafka::ConsumerOut::stat,
-											_kafkaConsumerCallbackOut,
-											std::placeholders::_1, std::placeholders::_2));
-		consumer->setKafkaDebug(_configure->kafkaDebugOut);
-		consumer->setOffsetStorePath(_configure->offsetPathOut);
-		consumer->setKafkaStatInterval(_configure->statIntervalOut);
-		if (_configure->isNewConsumerOut) {
-			consumer->setIsNewConsumer(true);
-			consumer->setOffsetStoreMethod("broker");
-		}
-		_kafkas[t] = consumer;
-	}
+    LOG_INFO << "Topic list:" << _configure->topicNameConsumerOut;
+
+    _kafka = new adbase::kafka::ConsumerBatch(topicNames, _configure->groupIdOut, 
+                                            _configure->brokerListConsumerOut);
+    _kafka->setMessageHandler(std::bind(&aims::kafka::ConsumerOut::pull,
+                _kafkaConsumerCallbackOut,
+                std::placeholders::_1, std::placeholders::_2,
+                std::placeholders::_3, std::placeholders::_4));
+    _kafka->setStatCallback(std::bind(&aims::kafka::ConsumerOut::stat,
+                _kafkaConsumerCallbackOut,
+                std::placeholders::_1, std::placeholders::_2));
+	_kafka->setKafkaDebug(_configure->kafkaDebugOut);
+	_kafka->setKafkaStatInterval(_configure->statIntervalOut);
 }
 
 // }}}
diff --git a/src/Aims.hpp b/src/Aims.hpp
@@ -12,6 +12,7 @@ class Aims {
 	Aims(AimsContext* context);
 	~Aims();
 	void run();
+	void reload();
 
 private:
 	/// 传输上下文指针
@@ -22,7 +23,7 @@ class Aims {
 	void stop();
 	void initKafkaConsumer();
 	aims::kafka::ConsumerOut* _kafkaConsumerCallbackOut = nullptr;
-	std::unordered_map<std::string, adbase::kafka::Consumer*> _kafkas;
+    adbase::kafka::ConsumerBatch* _kafka;
 
 };
 

diff --git a/src/Aims/Kafka/ConsumerOut.cpp b/src/Aims/Kafka/ConsumerOut.cpp
@@ -33,9 +33,8 @@ bool ConsumerOut::pull(const std::string& topicName, int partId, uint64_t offset
 // }}}
 // {{{ void ConsumerOut::stat()
 
-void ConsumerOut::stat(adbase::kafka::Consumer* consumer, const std::string& stats) {
+void ConsumerOut::stat(adbase::kafka::ConsumerBatch*, const std::string& stats) {
     LOG_INFO << "Stats:" << stats.substr(0, 1024);
-    LOG_INFO << consumer->getTopicName();
 }
 
 // }}}

diff --git a/src/Aims/Kafka/ConsumerOut.hpp b/src/Aims/Kafka/ConsumerOut.hpp
@@ -12,11 +12,11 @@ class ConsumerOut {
 	ConsumerOut(AimsContext* context);
 	~ConsumerOut();
 	bool pull(const std::string& topicName, int partId, uint64_t offset, const adbase::Buffer& data);
-	void stat(adbase::kafka::Consumer* consumer, const std::string& stats);
+	void stat(adbase::kafka::ConsumerBatch* consumer, const std::string& stats);
 private:
 	AimsContext* _context;
 };
 
 }
 }
-#endif
+#endif
diff --git a/src/App.cpp b/src/App.cpp
@@ -5,27 +5,18 @@
 
 //{{{ macros
 
-#define LOAD_KAFKA_CONSUMER_CONFIG(name, sectionName) do {\
-    _configure->isNewConsumer##name = config.getOptionBool("kafkac_"#sectionName, "isNewConsumer"#name);\
-    _configure->topicNameConsumer##name    = config.getOption("kafkac_"#sectionName, "topicName"#name);\
-    _configure->groupId##name      = config.getOption("kafkac_"#sectionName, "groupId"#name);\
-    _configure->brokerListConsumer##name   = config.getOption("kafkac_"#sectionName, "brokerList"#name);\
-    _configure->kafkaDebug##name   = config.getOption("kafkac_"#sectionName, "kafkaDebug"#name);\
-    _configure->offsetPath##name   = config.getOption("kafkac_"#sectionName, "offsetPath"#name);\
-    _configure->statInterval##name = config.getOption("kafkac_"#sectionName, "statInterval"#name);\
-} while(0)
-
-#define LOAD_KAFKA_PRODUCER_CONFIG(name, sectionName) do {\
-    _configure->topicNameProducer##name    = config.getOption("kafkap_"#sectionName, "topicName"#name);\
-    _configure->brokerListProducer##name   = config.getOption("kafkap_"#sectionName, "brokerList"#name);\
-    _configure->debug##name        = config.getOption("kafkap_"#sectionName, "debug"#name);\
-    _configure->queueLength##name  = config.getOptionUint32("kafkap_"#sectionName, "queueLength"#name);\
-} while(0)
-
 #define LOAD_TIMER_CONFIG(name) do {\
 	_configure->interval##name = config.getOptionUint32("timer", "interval"#name);\
 } while(0)
 
+#define LOAD_KAFKA_CONSUMER_CONFIG(name, sectionName) do {\
+        _configure->topicNameConsumer##name    = config.getOption("kafkac_"#sectionName, "topicName"#name);\
+        _configure->groupId##name      = config.getOption("kafkac_"#sectionName, "groupId"#name);\
+        _configure->brokerListConsumer##name   = config.getOption("kafkac_"#sectionName, "brokerList"#name);\
+        _configure->kafkaDebug##name   = config.getOption("kafkac_"#sectionName, "kafkaDebug"#name);\
+        _configure->statInterval##name = config.getOption("kafkac_"#sectionName, "statInterval"#name);\
+} while(0)
+
 //}}}
 // {{{ App::App()
 
@@ -46,7 +37,9 @@ void App::run() {
 	_storage = std::shared_ptr<app::Storage>(new app::Storage(_configure));
 	_storage->init();
 
-	_message = std::shared_ptr<app::Message>(new app::Message(_configure, _storage.get()));
+	_metrics = std::shared_ptr<app::Metrics>(new app::Metrics());
+
+	_message = std::shared_ptr<app::Message>(new app::Message(_configure, _storage.get(), _metrics.get()));
 	_message->start();
 }
 
@@ -71,6 +64,7 @@ void App::stop() {
 void App::setAdServerContext(AdServerContext* context) {
 	context->app = this;
 	context->storage = _storage.get();
+	context->appMetrics = _metrics.get();
 }
 
 // }}}
@@ -80,6 +74,7 @@ void App::setAimsContext(AimsContext* context) {
 	context->app = this;
 	context->message = _message.get();
 	context->storage = _storage.get();
+	context->appMetrics = _metrics.get();
 }
 
 // }}}
@@ -88,14 +83,13 @@ void App::setAimsContext(AimsContext* context) {
 void App::setTimerContext(TimerContext* context) {
 	context->app = this;
 	context->storage = _storage.get();
+	context->appMetrics = _metrics.get();
 }
 
 // }}}
 //{{{ void App::loadConfig()
 
 void App::loadConfig(adbase::IniConfig& config) {
-	LOAD_KAFKA_CONSUMER_CONFIG(Out, out);
-
     _configure->luaDebug = config.getOptionBool("lua", "debug");
     _configure->luaScriptPath = config.getOption("lua", "scriptPath");
     _configure->consumerScriptName  = config.getOption("consumer", "scriptName");
@@ -105,6 +99,7 @@ void App::loadConfig(adbase::IniConfig& config) {
     _configure->messageSwp = config.getOption("consumer", "messageSwp");
     _configure->httpScriptName  = config.getOption("http", "scriptName");
 
+    LOAD_KAFKA_CONSUMER_CONFIG(Out, out);
 	LOAD_TIMER_CONFIG(ClearStorage);
 }
 

diff --git a/src/App.hpp b/src/App.hpp
@@ -6,6 +6,7 @@
 #include "AdbaseConfig.hpp"
 #include "App/Message.hpp"
 #include "App/Storage.hpp"
+#include "App/Metrics.hpp"
 
 class App {
 public:
@@ -24,6 +25,7 @@ class App {
 	AdbaseConfig* _configure;
 	std::shared_ptr<app::Storage> _storage;
 	std::shared_ptr<app::Message> _message;
+	std::shared_ptr<app::Metrics> _metrics;
 	mutable std::mutex _mut;
 	void bindLuaMessage();
 };

diff --git a/src/App/Message.cpp b/src/App/Message.cpp
@@ -1,5 +1,6 @@
 #include "Message.hpp"
 #include "Storage.hpp"
+#include "Metrics.hpp"
 #include <fstream>
 #include <adbase/Lua.hpp>
 
@@ -9,7 +10,9 @@ thread_local std::unordered_map<std::string, MessageItem> messageLuaMessages;
 
 // {{{ Message::Message()
 
-Message::Message(AdbaseConfig* configure, Storage* storage): _configure(configure), _storage(storage) {
+Message::Message(AdbaseConfig* configure, Storage* storage, Metrics* metrics): _configure(configure), 
+    _storage(storage),
+    _metrics(metrics) {
 }
 
 // }}}
@@ -120,6 +123,7 @@ void Message::initLua() {
 	clazz.addMethod("rollback", rollbackFn);
 
 	_storage->bindClass(messageLuaEngine.get());
+	_metrics->bindClass(messageLuaEngine.get());
 }
 
 // }}}

diff --git a/src/App/Message.hpp b/src/App/Message.hpp
@@ -14,6 +14,7 @@
 
 namespace app {
 class Storage;
+class Metrics;
 // message queue
 typedef struct MessageItem {
     int partId;
@@ -29,7 +30,7 @@ typedef std::list<std::list<std::string>> MessageToLua;
 
 class Message {
 public:
-	Message(AdbaseConfig* configure, Storage* storage);
+	Message(AdbaseConfig* configure, Storage* storage, Metrics* metrics);
 	~Message();
 	void start();
 	void stop();
@@ -47,6 +48,7 @@ class Message {
 
 	AdbaseConfig *_configure;
 	Storage* _storage;
+	Metrics* _metrics;
 	std::unordered_map<int, MessageQueue*> _queues;
 
 	void callMessage();