編輯:關於android開發
2)運行flume agent
- agent.sources = seqGenSrc
- agent.channels = memoryChannel
- agent.sinks = loggerSink
- # For each one of the sources, the type is defined
- agent.sources.seqGenSrc.type = exec
- agent.sources.seqGenSrc.command = tail -F /data/mongodata/mongo.log
- #agent.sources.seqGenSrc.bind = 172.168.49.130
- # The channel can be defined as follows.
- agent.sources.seqGenSrc.channels = memoryChannel
- # Each sink's type must be defined
- agent.sinks.loggerSink.type = file_roll
- agent.sinks.loggerSink.sink.directory = /data/flume
- #Specify the channel the sink should use
- agent.sinks.loggerSink.channel = memoryChannel
- # Each channel's type is defined.
- agent.channels.memoryChannel.type = memory
- # Other config values specific to each type of channel(sink or source)
- # can be defined as well
- # In this case, it specifies the capacity of the memory channel
- agent.channels.memoryChannel.capacity = 1000
- agent.channels.memory4log.transactionCapacity = 100
下一步,修改flume配置文件,將其中sink部分的配置改成kafka sink,如:
- Sink.Status status = Status.READY;
- Channel ch = getChannel();
- Transaction transaction = null;
- Event event = null;
- String eventTopic = null;
- String eventKey = null;
- try {
- transaction = ch.getTransaction();
- transaction.begin();
- messageList.clear();
- if (type.equals("sync")) {
- event = ch.take();
- if (event != null) {
- byte[] tempBody = event.getBody();
- String eventBody = new String(tempBody,"UTF-8");
- Map<String, String> headers = event.getHeaders();
- if ((eventTopic = headers.get(TOPIC_HDR)) == null) {
- eventTopic = topic;
- }
- eventKey = headers.get(KEY_HDR);
- if (logger.isDebugEnabled()) {
- logger.debug("{Event} " + eventTopic + " : " + eventKey + " : "
- + eventBody);
- }
- ProducerData<String, Message> data = new ProducerData<String, Message>
- (eventTopic, new Message(tempBody));
- long startTime = System.nanoTime();
- logger.debug(eventTopic+"++++"+eventBody);
- producer.send(data);
- long endTime = System.nanoTime();
- }
- } else {
- long processedEvents = 0;
- for (; processedEvents < batchSize; processedEvents += 1) {
- event = ch.take();
- if (event == null) {
- break;
- }
- byte[] tempBody = event.getBody();
- String eventBody = new String(tempBody,"UTF-8");
- Map<String, String> headers = event.getHeaders();
- if ((eventTopic = headers.get(TOPIC_HDR)) == null) {
- eventTopic = topic;
- }
- eventKey = headers.get(KEY_HDR);
- if (logger.isDebugEnabled()) {
- logger.debug("{Event} " + eventTopic + " : " + eventKey + " : "
- + eventBody);
- logger.debug("event #{}", processedEvents);
- }
- // create a message and add to buffer
- ProducerData<String, String> data = new ProducerData<String, String>
- (eventTopic, eventBody);
- messageList.add(data);
- }
- // publish batch and commit.
- if (processedEvents > 0) {
- long startTime = System.nanoTime();
- long endTime = System.nanoTime();
- }
- }
- transaction.commit();
- } catch (Exception ex) {
- String errorMsg = "Failed to publish events";
- logger.error("Failed to publish events", ex);
- status = Status.BACKOFF;
- if (transaction != null) {
- try {
- transaction.rollback();
- } catch (Exception e) {
- logger.error("Transaction rollback failed", e);
- throw Throwables.propagate(e);
- }
- }
- throw new EventDeliveryException(errorMsg, ex);
- } finally {
- if (transaction != null) {
- transaction.close();
- }
- }
- return status;
type指向kafkasink所在的完整路徑
- producer.sinks.r.type = org.apache.flume.sink.kafka.KafkaSink
- producer.sinks.r.brokerList = bigdata-node00:9092
- producer.sinks.r.requiredAcks = 1
- producer.sinks.r.batchSize = 100
- #producer.sinks.r.kafka.producer.type=async
- #producer.sinks.r.kafka.customer.encoding=UTF-8
- producer.sinks.r.topic = testFlume1
ListView與Adapter筆記:ZrcListView github:https://github.com/zarics/ZrcListView 先
Android ClearEditText:輸入用戶名、密碼錯誤時整體刪除及輸入為空時候晃動提示,android官網 1 package com.lixu.cleared
MAT使用入門,mat入門原文出處: 高建武 (Granker,@高爺) MAT簡介 MAT(Memory Analyzer
讓TextView出現跑馬燈效果,textview跑馬燈效果 只需要在TextView中添加一些屬性即可: <?xml version=1.0 encoding=