From 9964bb75b19fdbe7ea0e004d38ee6b2735b839c0 Mon Sep 17 00:00:00 2001
From: rewerma <rewerma@163.com>
Date: Wed, 13 Jun 2018 14:01:30 +0800
Subject: [PATCH] =?UTF-8?q?kafka=20=E5=AE=A2=E6=88=B7=E7=AB=AF=20example?=
 =?UTF-8?q?=20=E5=AE=8C=E5=96=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../running/KafkaCanalClientExample.java      | 64 ++++++++++---------
 1 file changed, 33 insertions(+), 31 deletions(-)

diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java
index 0f155c35..4527d847 100644
--- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java
+++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java
@@ -86,40 +86,42 @@ public class KafkaCanalClientExample {
     }
 
     private void process() {
-        while (running) {
-            try {
-                connector.subscribe();
-                while (running) {
-                    try {
-                        Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); //获取message
-                        if (message == null) {
-                            continue;
-                        }
-                        long batchId = message.getId();
-                        int size = message.getEntries().size();
-                        if (batchId == -1 || size == 0) {
-                            // try {
-                            // Thread.sleep(1000);
-                            // } catch (InterruptedException e) {
-                            // }
-                        } else {
-                            // printSummary(message, batchId, size);
-                            // printEntry(message.getEntries());
-                            logger.info(message.toString());
-                        }
-
-                        connector.ack(); // 提交确认
-                    } catch (WakeupException e) {
-                        //ignore
-                    } catch (Exception e) {
-                        logger.error(e.getMessage(), e);
+        try {
+            connector.subscribe();
+            while (running) {
+                try {
+                    Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); //获取message
+                    if (message == null) {
+                        continue;
                     }
+                    long batchId = message.getId();
+                    int size = message.getEntries().size();
+                    if (batchId == -1 || size == 0) {
+                        // try {
+                        // Thread.sleep(1000);
+                        // } catch (InterruptedException e) {
+                        // }
+                    } else {
+                        // printSummary(message, batchId, size);
+                        // printEntry(message.getEntries());
+                        logger.info(message.toString());
+                    }
+
+                    connector.ack(); // 提交确认
+                } catch (WakeupException e) {
+                    try {
+                        Thread.sleep(500); //延时确保running状态的改变
+                    } catch (InterruptedException ie) {
+                        //ignore
+                    }
+                } catch (Exception e) {
+                    logger.error(e.getMessage(), e);
                 }
-            } catch (WakeupException e) {
-                //ignore
-            } catch (Exception e) {
-                logger.error(e.getMessage(), e);
             }
+        } catch (WakeupException e) {
+            //ignore
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
         }
     }
 }
\ No newline at end of file