-
Notifications
You must be signed in to change notification settings - Fork 14
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
2a95912
commit c4965c1
Showing
16 changed files
with
352 additions
and
25 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| server: "localhost:9092" | ||
| topic: "mo-stream-test" | ||
| client_id: "mo-kafka-client" |
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| package io.mo.stream; | ||
|
|
||
| import org.apache.kafka.clients.consumer.KafkaConsumer; | ||
|
|
||
| public class Consumer { | ||
| KafkaConsumer<String,String> consumer; | ||
|
|
||
| public Consumer(KafkaConsumer consumer){ | ||
| this.consumer = consumer; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| package io.mo.stream; | ||
|
|
||
| import io.mo.util.KafkaConfiUtil; | ||
| import org.apache.kafka.clients.producer.KafkaProducer; | ||
| import org.apache.kafka.clients.producer.ProducerConfig; | ||
| import org.apache.kafka.common.serialization.StringSerializer; | ||
|
|
||
| import java.util.Properties; | ||
|
|
||
| public class KafkaManager { | ||
|
|
||
| public static Producer getProducer(){ | ||
| Properties properties = new Properties(); | ||
| properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfiUtil.getServerAddr()); | ||
| properties.put(ProducerConfig.CLIENT_ID_CONFIG,KafkaConfiUtil.getClientId()); | ||
| properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); | ||
| properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); | ||
|
|
||
| KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties); | ||
| Producer producer = new Producer(kafkaProducer); | ||
| return producer; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package io.mo.stream; | ||
|
|
||
| import org.apache.kafka.clients.producer.KafkaProducer; | ||
| import org.apache.kafka.clients.producer.ProducerRecord; | ||
|
|
||
| import java.util.concurrent.ExecutionException; | ||
|
|
||
| public class Producer { | ||
| private KafkaProducer<String,String> producer; | ||
|
|
||
| public Producer(KafkaProducer producer){ | ||
| this.producer = producer; | ||
| } | ||
|
|
||
| public boolean send(String topic, String message){ | ||
|
|
||
| ProducerRecord<String,String> record = new ProducerRecord<String,String>(topic,message); | ||
| try { | ||
| producer.send(record).get(); | ||
| } catch (InterruptedException e) { | ||
| e.printStackTrace(); | ||
| return false; | ||
| } catch (ExecutionException e) { | ||
| e.printStackTrace(); | ||
| return false; | ||
| } | ||
| producer.close(); | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
| public boolean send(TopicAndRecords tar){ | ||
| for(int i = 0; i < tar.size(); i++){ | ||
| try { | ||
| ProducerRecord<String,String> record = new ProducerRecord<String,String>(tar.getTopic(),tar.getRecord(i)); | ||
| producer.send(record).get(); | ||
| } catch (InterruptedException e) { | ||
| e.printStackTrace(); | ||
| return false; | ||
| } catch (ExecutionException e) { | ||
| e.printStackTrace(); | ||
| return false; | ||
| } | ||
| } | ||
|
|
||
| producer.close(); | ||
| return true; | ||
| } | ||
|
|
||
| public void close(){ | ||
| this.producer.close(); | ||
| } | ||
|
|
||
|
|
||
| public static void main(String[] args){ | ||
| Producer producer = KafkaManager.getProducer(); | ||
| producer.send("mo-stream-test","First message to topic mo-stream-test"); | ||
|
|
||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| package io.mo.stream; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| public class TopicAndRecords { | ||
| private String topic; | ||
| private List<String> records = new ArrayList<>(); | ||
|
|
||
| public TopicAndRecords(){ | ||
|
|
||
| } | ||
|
|
||
| public TopicAndRecords(String topic){ | ||
| this.topic = topic; | ||
| } | ||
|
|
||
| public void addRecord(String record){ | ||
| records.add(record); | ||
| } | ||
|
|
||
| public int size(){ | ||
| return records.size(); | ||
| } | ||
|
|
||
| public String getRecord(int i){ | ||
| return records.get(i); | ||
| } | ||
|
|
||
| public String getTopic(){ | ||
| return topic; | ||
| } | ||
|
|
||
| public void setTopic(String topic){ | ||
| this.topic = topic; | ||
| } | ||
|
|
||
| public String[] getRecords(){ | ||
| if(records.size() > 0){ | ||
| return records.toArray(new String[records.size()]); | ||
| } | ||
|
|
||
| return null; | ||
| } | ||
|
|
||
| public String getRecordsStr(){ | ||
| StringBuffer buffer = new StringBuffer(); | ||
| for(int i = 0; i < records.size();i++){ | ||
| buffer.append(records.get(i)); | ||
| buffer.append("\n"); | ||
| } | ||
|
|
||
| if(buffer.length() > 0 ) | ||
| return buffer.toString(); | ||
| return null; | ||
| } | ||
| } |
Oops, something went wrong.