Golang logger.Warning函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Golang中github.com/dzch/go-utils/logger.Warning函数的典型用法代码示例。如果您正苦于以下问题：Golang Warning函数的具体用法？Golang Warning怎么用？Golang Warning使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。

在下文中一共展示了Warning函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Golang代码示例。

示例1: transData

func (hp *HttpProtocol) transData(server string, transData *TransData) (err error) {
	/* compose req */
	uri := hp.regTopic.ReplaceAllString(hp.uri, transData.topic)
	uri = hp.regMethod.ReplaceAllString(uri, transData.method)
	uri = hp.regPartition.ReplaceAllString(uri, strconv.FormatInt(int64(transData.partition), 10))
	uri = hp.regTransid.ReplaceAllString(uri, strconv.FormatInt(transData.transid, 10))
	url := fmt.Sprintf("http://%s%s", server, uri)
	req, err := http.NewRequest("POST", url, bytes.NewReader(transData.data))
	if err != nil {
		logger.Warning("module [%s]: fail to transData: url=%s, topic=%s, partition=%d, transid=%d, method=%s, err=%s", hp.moduleName, url, transData.topic, transData.partition, transData.transid, transData.method, err.Error())
		return err
	}
	/* add header */
	req.Header = hp.header
	req.Host = hp.reqHeaderHost
	/* Post */
	res, err := hp.client.Do(req)
	if err != nil {
		logger.Warning("module [%s]: fail to transData: url=%s, topic=%s, partition=%d, transid=%d, method=%s, err=%s", hp.moduleName, url, transData.topic, transData.partition, transData.transid, transData.method, err.Error())
		return err
	}
	defer res.Body.Close()
	/* check res: 200 不重试；其他，重试 */
	if res.StatusCode == http.StatusOK {
		logger.Notice("module [%s]: success transData: url=%s, topic=%s, partition=%d, transid=%d, method=%s, datalen=%d, http_status_code=%d", hp.moduleName, url, transData.topic, transData.partition, transData.transid, transData.method, len(transData.data), res.StatusCode)
		return nil
	} else {
		logger.Warning("module [%s]: fail to transData: url=%s, topic=%s, partition=%d, transid=%d, method=%s, http_status_code=%d", hp.moduleName, url, transData.topic, transData.partition, transData.transid, transData.method, res.StatusCode)
		return errors.New("fail to trans")
	}
	return nil
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:32，代码来源:http_protocol.go

示例2: updateBrokerListAndWatch

func (z *Zookeeper) updateBrokerListAndWatch() {
	for {
		children, _, eventChan, err := z.zkConn.ChildrenW(fmt.Sprintf("%s/brokers/ids", z.cc.chroot))
		if err != nil {
			err = errors.New(fmt.Sprintf("fail to zk.ChildrenW: %s", err.Error()))
			logger.Fatal(err.Error())
			z.fatalErrorChan <- err
			return
		}
		if len(children) == 0 {
			logger.Warning("no broker found")
		} else {
			logger.Debug("using child: %s", children[0])
			val, _, err := z.zkConn.Get(fmt.Sprintf("%s/brokers/ids/%s", z.cc.chroot, children[0]))
			var brokerIdVal BrokerIdVal
			err = json.Unmarshal(val, &brokerIdVal)
			if err != nil {
				logger.Warning("fail to json.Unmarshal for broker id %s: %s", children[0], err.Error())
			}
			logger.Debug("broker id %s: host [%s] port [%d]", children[0], brokerIdVal.Host, brokerIdVal.Port)
			z.brokerIdValChan <- &brokerIdVal
		}
		<-eventChan
	}
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:25，代码来源:zookeeper.go

示例3: ServeHTTP

func (h *GetHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	startTime := time.Now()
	qv := r.URL.Query()
	id, partition, offset, err := h.bs.km.parseKey(qv.Get("key"))
	if err != nil {
		logger.Warning("fail to parseKey: %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	val, err := h.getData(id, partition, offset)
	if err != nil {
		logger.Warning("fail to getData: %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusOK)
	_, err = w.Write(val)
	if err != nil {
		logger.Warning("fail to write response: %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	endTime := time.Now()
	costTimeUS := endTime.Sub(startTime) / time.Microsecond
	logger.Notice("success process get: %s, cost_us=%d, datalen=%d", r.URL.String(), costTimeUS, len(val))
	return
}

开发者ID:dzch，项目名称:binstore，代码行数:27，代码来源:get_handler.go

示例4: ServeHTTP

func (h *StoreHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	startTime := time.Now()
	/* check query */
	if r.ContentLength <= 0 {
		logger.Warning("invalid query, need post data: %s", r.URL.String())
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	// qv := r.URL.Query()
	sr := h.getStoreReq()
	defer h.putStoreReq(sr)
	nr, err := sr.reqBuffer.ReadFrom(r.Body)
	if int64(nr) != r.ContentLength || err != nil {
		logger.Warning("fail to read body: %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	err = h.parseReq(sr)
	if err != nil {
		logger.Warning("invalid query, parseReq failed : %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	err = h.bs.store.addNewData(sr)
	if err != nil {
		logger.Warning("fail to write response: %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusOK)
	endTime := time.Now()
	costTimeUS := endTime.Sub(startTime) / time.Microsecond
	logger.Notice("success process store: %s, id=%d, cost_us=%d", r.URL.String(), sr.id, costTimeUS)
	return
}

开发者ID:dzch，项目名称:binstore，代码行数:35，代码来源:store_handler.go

示例5: updateBrokerOffset

func (b *Broker) updateBrokerOffset() {
	if b.client == nil || b.client.Closed() {
		logger.Notice("broker.client not ready yet")
		return
	}
	topics, err := b.client.Topics()
	if err != nil {
		logger.Warning("fail to b.client.Topics(): %s", err.Error())
		return
	}
	for _, topic := range topics {
		partitions, err := b.client.Partitions(topic)
		if err != nil {
			logger.Warning("fail to b.client.Partitions(): %s", err.Error())
			continue
		}
		for _, partition := range partitions {
			offset, err := b.client.GetOffset(topic, partition, sarama.OffsetNewest)
			if err != nil {
				logger.Warning("fail to b.client.GetOffset topic=%s, partition=%d, err=%s", topic, partition, err.Error())
				continue
			}
			// TODO: use pool
			b.brokerOffsetChan <- &BrokerOffset{
				topic:     topic,
				partition: partition,
				offset:    offset,
			}
			logger.Debug("offset: topic=%s, partition=%d, offset=%d", topic, partition, offset)
		}
	}
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:32，代码来源:broker.go

示例6: ServeHTTP

func (ch *CmHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	startTime := time.Now()
	/* check query */
	if r.ContentLength <= 0 {
		logger.Warning("invalid query, need post data: %s", r.URL.String())
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	qv := r.URL.Query()
	post := make([]byte, r.ContentLength)
	nr, err := io.ReadFull(r.Body, post)
	if int64(nr) != r.ContentLength {
		logger.Warning("fail to read body: %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	/* compose CmData */
	cmData := ch.cdp.fetch()
	defer ch.cdp.put(cmData)
	switch qv.Get("acks") {
	case "0":
		cmData.requiredAcks = sarama.NoResponse
	case "1":
		cmData.requiredAcks = sarama.WaitForLocal
	case "-1":
		cmData.requiredAcks = sarama.WaitForAll
	default:
		cmData.requiredAcks = sarama.WaitForLocal
	}
	cmData.topic = qv.Get("topic")
	cmData.key = qv.Get("key")
	cmData.data = post
	/* commit */
	ch.kp.producer.produce(cmData)
	/* wait res */
	<-cmData.cmDoneChan
	if cmData.err != nil {
		logger.Warning("fail to commit req: %s, error: %s", r.URL.String(), (*(cmData.err)).Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	endTime := time.Now()
	costTimeUS := endTime.Sub(startTime) / time.Microsecond
	// TODO
	logger.Notice("success process commit: %s, cost_us=%d, datalen=%d, offset=%d, partition=%d", r.URL.String(), costTimeUS, nr, cmData.offset, cmData.partition)
	w.WriteHeader(http.StatusOK)
	return
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:48，代码来源:cm_handler.go

示例7: computeTransDelay

func (c *Cluster) computeTransDelay(transDelayChan chan TransDelay) {
	resMap := make(map[string]map[string]int64)
	for cg, tm := range c.cs.transfer.transferOffset {
		for topic, pm := range tm {
			dpm, ok := resMap[cg]
			if !ok {
				dpm = make(map[string]int64)
				resMap[cg] = dpm
			}
			for partition, tor := range pm {
				bor, ok := c.cs.broker.brokerOffset[topic][partition]
				if !ok {
					logger.Warning("no broker offset record found: topic=%s, partition=%d", topic, partition)
					dpm[strconv.FormatInt(int64(partition), 10)] = 0
					continue
				}
				bo := c.computeBrokerOffsetAtTime(tor.last.timestamp, bor)
				delay := bo - tor.last.offset
				if delay < 0 {
					delay = 0
				}
				dpm[strconv.FormatInt(int64(partition), 10)] = delay
			}
		}
	}
	transDelayChan <- TransDelay(resMap)
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:27，代码来源:cluster.go

示例8: processAckData

/* do not change state in this func */
func (td *TransDi) processAckData(ackData *AckData) {
	transid := ackData.transid
	workerId := ackData.workerId
	logger.Debug("process ack: transid=%d, workerId=%d", transid, workerId)
	/* narrow window */
	msg := td.transWindow.ackOne(transid)
	if msg != nil && msg.Offset > td.ackedMinTransid {
		td.ackedMinTransid = msg.Offset
		err := td.cg.CommitUpto(msg)
		if err != nil {
			// TODO: optimized
			logger.Warning("fail to consumergroup.CommitUpto(): %s", err.Error())
			td.fatalErrorChan <- err
			return
		}
		logger.Debug("consumergroup.CommitUpTo %d", msg.Offset)
	}
	logger.Debug("transWindow size: %d", td.transWindow.window.Len())
	/* move worker */
	worker := td.transWorkers[workerId]
	worker.inWork = false
	err := worker.workIfNeed()
	if err != nil {
		logger.Fatal("fail to let worker to work: %s", err.Error())
		td.fatalErrorChan <- err
		return
	}
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:29，代码来源:transdi.go

示例9: updateCgAndWatch

func (z *Zookeeper) updateCgAndWatch(cg string) {
	path := fmt.Sprintf("%s/consumers/%s/offsets", z.cc.chroot, cg)
	for {
		children, _, eventChan, err := z.zkConn.ChildrenW(path)
		if err == zk.ErrNoNode {
			_, _, eventChan, err = z.zkConn.ExistsW(path)
			if err != nil {
				err = errors.New(fmt.Sprintf("fail to zk.ExistsW: path=%s, %s", path, err.Error()))
				logger.Fatal(err.Error())
				z.fatalErrorChan <- err
				return
			}
			<-eventChan
			continue
		}
		if err != nil {
			err = errors.New(fmt.Sprintf("fail to zk.ChildrenW: path=%s, %s", path, err.Error()))
			logger.Fatal(err.Error())
			z.fatalErrorChan <- err
			return
		}
		if len(children) == 0 {
			logger.Warning("no transfer topic found: path=%s", path)
		} else {
			for _, topic := range children {
				go z.updateTopicAndWatch(cg, topic)
			}
		}
		event := <-eventChan
		// let root watcher to re-init
		if event.Type == zk.EventNotWatching {
			return
		}
	}
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:35，代码来源:zookeeper.go

示例10: Run

func (d *Deleter) Run() (err error) {
	err = d.httpServer.ListenAndServe()
	if err != nil {
		logger.Warning("fail to httpServer.ListenAndServe: %s", err.Error())
		return
	}
	logger.Notice("httpServer.ListenAndServe end")
	return nil
}

开发者ID:dzch，项目名称:cachedeleter，代码行数:9，代码来源:deleter.go

示例11: fetch

func (pmp *PMsgPool) fetch() *PMsgPoolEle {
	if pmp.poolFreeSize <= 0 {
		logger.Warning("Producer Message Pool is too small")
		return pmp.newPMsgPoolEle()
	}
	r := pmp.root.r.Unlink(1)
	pmp.poolFreeSize--
	return r.Value.(*PMsgPoolEle)
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:9，代码来源:producer_msg_pool.go

示例12: fetch

func (cdp *CmDataPool) fetch() *CmData {
	var cd *CmData
	select {
	case cd = <-cdp.pool:
	default:
		logger.Warning("cm_data_pool is not big enough")
		cd = cdp.newCmData()
	}
	return cd
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:10，代码来源:cm_data_pool.go

示例13: run

func (zk *ZK) run() {
	for {
		err := zk.updateOffsets()
		if err != nil {
			logger.Warning("fail to zk.updateOffsets: %s", err.Error())
			time.Sleep(zk.config.zkFailRetryInterval)
		}
		time.Sleep(zk.config.zkUpdateInterval)
	}
}

开发者ID:dzch，项目名称:binstore，代码行数:10，代码来源:zookeeper.go

示例14: initProtocol

func (worker *TransWorker) initProtocol() (err error) {
	name, ok := worker.protocolConfig["name"]
	if !ok {
		err = errors.New(fmt.Sprintf("protocol.name not found in module conf file, module [%s]", worker.moduleName))
		logger.Warning("%s", err.Error())
		return
	}
	worker.protocolName = name.(string)
	switch worker.protocolName {
	case "http":
		err = worker.initHttpProtocol()
	default:
		err = errors.New(fmt.Sprintf("unknown protocol: %s", worker.protocolName))
	}
	if err != nil {
		logger.Warning("fail to init protocol for module [%s]: %s", worker.moduleName, err.Error())
		return
	}
	return nil
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:20，代码来源:worker.go

示例15: handleCm

func (httpServer *HttpServer) handleCm(w http.ResponseWriter, r *http.Request) {
	startTime := time.Now()
	/* check query */
	if r.ContentLength <= 0 {
		logger.Warning("invalid query, need post data: %s", r.URL.String())
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	qv := r.URL.Query()
	post := make([]byte, r.ContentLength)
	nr, err := io.ReadFull(r.Body, post)
	if int64(nr) != r.ContentLength {
		logger.Warning("fail to read body: %s, %s", r.URL.String(), err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	/* compose CmData */
	cmData := httpServer.cdp.fetch()
	defer httpServer.cdp.put(cmData)
	cmData.topic = qv.Get("topic")
	cmData.key = qv.Get("key")
	cmData.data = post
	/* commit */
	httpServer.cmDataChan <- cmData
	/* wait res */
	<-cmData.cmDoneChan
	if cmData.err != nil {
		logger.Warning("fail to commit req: %s, error: %s", r.URL.String(), (*(cmData.err)).Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	endTime := time.Now()
	costTimeUS := endTime.Sub(startTime) / time.Microsecond
	// TODO
	logger.Notice("success process commit: %s, cost_us=%d, datalen=%d, offset=%d, partition=%d", r.URL.String(), costTimeUS, nr, cmData.offset, cmData.partition)
	w.WriteHeader(http.StatusOK)
	return
}

开发者ID:dzch，项目名称:kproxy，代码行数:38，代码来源:http_server.go

示例16: initRegs

func (hp *HttpProtocol) initRegs() (err error) {
	hp.regTopic, err = regexp.Compile("{#TOPIC}")
	if err != nil {
		logger.Warning("fail to regexp.Compile: %s", err.Error())
		return
	}
	hp.regMethod, err = regexp.Compile("{#METHOD}")
	if err != nil {
		logger.Warning("fail to regexp.Compile: %s", err.Error())
		return
	}
	hp.regPartition, err = regexp.Compile("{#PARTITION}")
	if err != nil {
		logger.Warning("fail to regexp.Compile: %s", err.Error())
		return
	}
	hp.regTransid, err = regexp.Compile("{#TRANSID}")
	if err != nil {
		logger.Warning("fail to regexp.Compile: %s", err.Error())
		return
	}
	return nil
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:23，代码来源:http_protocol.go

示例17: response

func (d *Deleter) response(w http.ResponseWriter, statusCode int, errMsg string) {
	var buf bytes.Buffer
	buf.Reset()
	res := map[string]interface{}{
		"error": map[string]interface{}{
			"errno": statusCode,
			/* some msgpack lib cannot support new encoding-spec， so let len(string) < 31 */
			//			"errmsg": "success process: /del?topic=ap_",
			"errmsg": "see errno please",
		},
	}
	wr := msgp.NewWriter(&buf)
	err := wr.WriteIntf(res)
	if err != nil {
		logger.Warning("fail to msgp.wr.WriteIntf: %s", err.Error())
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	wr.Flush()
	w.WriteHeader(statusCode)
	_, err = w.Write(buf.Bytes())
	if err != nil {
		logger.Warning("fail to http.ResponseWriter.Write(): %s", err.Error())
		return
	}
	//	/* unpack test */
	//	b := bytes.NewReader(buf.Bytes())
	//	b.Seek(0, 0)
	//	msgr := msgp.NewReader(b)
	//	reqi, err := msgr.ReadIntf()
	//	if err != nil {
	//        msg := fmt.Sprintf("fail to decode post data: %s", err.Error())
	//		logger.Warning("%s", msg)
	//		return
	//	}
	//	logger.Warning("unpack: %v", reqi)
}

开发者ID:dzch，项目名称:cachedeleter，代码行数:37，代码来源:deleter.go

示例18: updatePartitionAndWatch

func (z *Zookeeper) updatePartitionAndWatch(cg, topic string, partition int32) {
	path := fmt.Sprintf("%s/consumers/%s/offsets/%s/%d", z.cc.chroot, cg, topic, partition)
	for {
		val, _, eventChan, err := z.zkConn.GetW(path)
		if err == zk.ErrNoNode {
			_, _, eventChan, err = z.zkConn.ExistsW(path)
			if err != nil {
				err = errors.New(fmt.Sprintf("fail to zk.ExistsW: path=%s, %s", path, err.Error()))
				logger.Fatal(err.Error())
				z.fatalErrorChan <- err
				return
			}
			<-eventChan
			continue
		}
		if err != nil {
			err = errors.New(fmt.Sprintf("fail to zk.GetW: path=%s, %s", path, err.Error()))
			logger.Fatal(err.Error())
			z.fatalErrorChan <- err
			return
		}
		if len(val) == 0 {
			logger.Warning("transfer partition val is empty: path=%s", path)
		} else {
			offset, err := strconv.ParseInt(string(val), 10, 64)
			if err != nil {
				err = errors.New(fmt.Sprintf("fail to transfer: strconv.ParseInt: %s", err.Error()))
				logger.Fatal(err.Error())
				z.fatalErrorChan <- err
				return
			}
			z.transferOffsetChan <- &TransferOffset{
				cg:        cg,
				topic:     topic,
				partition: partition,
				offset:    offset,
			}
			logger.Debug("see transfer: cg=%s, topic=%s, partition=%d, offset=%d", cg, topic, partition, offset)
		}
		event := <-eventChan
		// let root watcher to re-init
		if event.Type == zk.EventNotWatching {
			return
		}
	}
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:46，代码来源:zookeeper.go

示例19: updateTransferAndWatch

func (z *Zookeeper) updateTransferAndWatch() {
	path := fmt.Sprintf("%s/consumers", z.cc.chroot)
	for {
		children, _, eventChan, err := z.zkConn.ChildrenW(path)
		if err != nil {
			err = errors.New(fmt.Sprintf("fail to zk.ChildrenW: path=%s, %s", path, err.Error()))
			logger.Fatal(err.Error())
			z.fatalErrorChan <- err
			return
		}
		if len(children) == 0 {
			logger.Warning("no transfer cg found: path=%s", path)
		} else {
			for _, cg := range children {
				go z.updateCgAndWatch(cg)
			}
		}
		<-eventChan
	}
}

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:20，代码来源:zookeeper.go

示例20: initConfig

func (m *Module) initConfig() (err error) {
	content, err := ioutil.ReadFile(fmt.Sprintf("%s/%s.yaml", m.moduleConfDir, m.name))
	if err != nil {
		logger.Warning("fail to read module config: %s", err.Error())
		return
	}
	mc := make(map[interface{}]interface{})
	err = yaml.Unmarshal(content, &mc)
	if err != nil {
		return
	}
	m.moduleConfig = mc

	/* topic conf */
	topic, ok := mc["topic"]
	if !ok {
		err = errors.New(fmt.Sprintf("topic not found in module conf file, module [%s]", m.name))
		logger.Warning("%s", err.Error())
		return
	}
	m.topic = topic.(string)

	/* method conf */
	m.methodEnabled = make(map[string]bool)
	methods, ok := mc["methods"]
	if ok {
		for _, method := range methods.([]interface{}) {
			m.methodEnabled[method.(string)] = true
		}
	}

	/* protocol conf */
	protocol, ok := mc["protocol"]
	if !ok {
		err = errors.New(fmt.Sprintf("protocol not found in module conf file, module [%s]", m.name))
		logger.Warning("%s", err.Error())
		return
	}
	m.protocolConfig = protocol.(map[interface{}]interface{})

	/* worker conf */
	workerNum, ok := mc["worker_num"]
	if !ok {
		err = errors.New(fmt.Sprintf("worker_num not found in module conf file, module [%s]", m.name))
		logger.Warning("%s", err.Error())
		return
	}
	m.workerNum = uint32(workerNum.(int))
	maxRetryTimes, ok := mc["max_retry_times"]
	if !ok {
		err = errors.New(fmt.Sprintf("max_retry_times not found in module conf file, module [%s]", m.name))
		logger.Warning("%s", err.Error())
		return
	}
	m.maxRetryTimes = maxRetryTimes.(int)
	failRetryInterval, ok := mc["fail_retry_interval_ms"]
	if !ok {
		err = errors.New(fmt.Sprintf("fail_retry_interval_ms not in module conf file, module [%s]", m.name))
		logger.Warning("%s", err.Error())
		return
	}
	m.failRetryInterval = time.Duration(failRetryInterval.(int)) * time.Millisecond
	sbykey, ok := mc["serialize_by_key"]
	if ok {
		m.serializeByKey = sbykey.(bool)
	} else {
		m.serializeByKey = true
	}

	/* window size */
	windowSize, ok := mc["window_size"]
	if !ok {
		err = errors.New(fmt.Sprintf("window_size not found in module conf file, module [%s]", m.name))
		logger.Warning("%s", err.Error())
		return
	}
	m.windowSize = windowSize.(int)

	/* backend servers */
	backendServers, ok := mc["backend_servers"]
	if !ok {
		err = errors.New(fmt.Sprintf("backend_servers not found in module conf file, module [%s]", m.name))
		logger.Warning("%s", err.Error())
		return
	}
	for _, server := range backendServers.([]interface{}) {
		m.backendServers = append(m.backendServers, server.(string))
	}

	/* consumer config */
	ti, ok := mc["expected_processing_time_ms"]
	if ok {
		m.expectedProcessingTime = time.Duration(ti.(int)) * time.Millisecond
	}
	ti, ok = mc["zk_offset_update_interval_sec"]
	if ok {
		m.zkOffsetUpdateInterval = time.Duration(ti.(int)) * time.Second
	}
	qs, ok := mc["waiting_queue_size"]
	if ok {
//.........这里部分代码省略.........

开发者ID:dzch，项目名称:kafka-http-wrapper，代码行数:101，代码来源:module.go

注：本文中的github.com/dzch/go-utils/logger.Warning函数示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。