Flink 数据源拆解分析 (WikipediaEditsSource)

Wikipedia provides an IRC channel where all edits to the wiki are logged.

  //远程连接的域名  public static final String DEFAULT_HOST = "irc.wikimedia.org";  //远程连接的端口  public static final int DEFAULT_PORT = 6667;  //IRC协议的channel  public static final String DEFAULT_CHANNEL = "#en.wikipedia";
  private final String host;  private final int port;  private final String channel;
  public WikipediaEditsSource() {    this(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_CHANNEL);  }
  public WikipediaEditsSource(String host, int port, String channel) {    this.host = host;    this.port = port;    this.channel = Objects.requireNonNull(channel);  }

  @Override  public void run(SourceContext<WikipediaEditEvent> ctx) throws Exception {    try (WikipediaEditEventIrcStream ircStream = new WikipediaEditEventIrcStream(host, port)) {      // 创建一个IRC协议的连接      ircStream.connect();      //进入指定的channel      ircStream.join(channel);
      try {        while (isRunning) {          //从阻塞队列中获取数据          WikipediaEditEvent edit = ircStream.getEdits().poll(100, TimeUnit.MILLISECONDS);          //如果取到了数据，就调用ctx.collect方法，将数据生产到Flink环境，给其他operator使用          if (edit != null) {            ctx.collect(edit);          }        }      } finally {        //结束时要向服务器发送数据表示离开        ircStream.leave(channel);      }    }  }

public void connect() throws IOException {    if (level != 0) // otherwise disconnected or connect      throw new SocketException("Socket closed or already open ("+ level +")");    IOException exception = null;    Socket s = null;    for (int i = 0; i < ports.length && s == null; i++) {      try {        //建立的是普通Socket连接        s = new Socket(host, ports[i]);        exception = null;      } catch (IOException exc) {        if (s != null)          s.close();        s = null;        exception = exc;      }    }    if (exception != null)      throw exception; // connection wasn't successful at any port
    prepare(s);  }

protected void prepare(Socket s) throws IOException {    if (s == null)      throw new SocketException("Socket s is null, not connected");    socket = s;    level = 1;    s.setSoTimeout(timeout);    in  = new BufferedReader(new InputStreamReader(s.getInputStream(),        encoding));    out = new PrintWriter(new OutputStreamWriter(s.getOutputStream(),        encoding));
    //IRCConnection是Thread的子类，执行start方法就表明会启动一个线程来执行IRCConnection的run方法    start();    //遵守IRC协议约定，发送一些注册相关的内容    register();  }

public void run() {    try {      String line;      while (!isInterrupted()) {        line = in.readLine();        if (line != null)          get(line);        else          close();      }    } catch (IOException exc) {      close();    }  }

//每当有人编辑了维基百科，这里就会收到一条command为PRIVMSG的记录if (command.equalsIgnoreCase("PRIVMSG")) { // MESSAGE      IRCUser user = p.getUser();      String middle = p.getMiddle();      String trailing = p.getTrailing();      for (int i = listeners.length - 1; i >= 0; i--)        //调用listener的onPrivmsg方法        listeners[i].onPrivmsg(middle, user, trailing);    }

@Overridepublic void onPrivmsg(String target, IRCUser user, String msg) {  LOG.debug("[{}] {}: {}.", target, user.getNick(), msg);  //根据消息构造一个WikipediaEditEvent对象，就是Flink的业务流程中用到的数据对象  WikipediaEditEvent event = WikipediaEditEvent.fromRawEvent(    System.currentTimeMillis(),    target,    msg);
  if (event != null) {    //eidts是个阻塞队列，WikipediaEditEvent被放入队列    if (!edits.offer(event)) {      LOG.debug("Dropping message, because of full queue.");    }  }}

while (isRunning) {    //从阻塞队列中获取数据    WikipediaEditEvent edit = ircStream.getEdits().poll(100, TimeUnit.MILLISECONDS);    //如果取到了数据，就调用ctx.collect方法，将数据生产到Flink环境，给其他operator使用    if (edit != null) {      ctx.collect(edit);    }}