Flink 自定义source、sink 是如何起作用的

......

	// flink 提供的 Rich*Function 系列算子的 open 和 close 方法被执行的地方
	@Override
	public void open() throws Exception {
		super.open();
//关键性方法 负责执行我们重写的open方法
		FunctionUtils.openFunction(userFunction, new Configuration());
	}

//关键性方法 负责执行我们重写的close方法
	@Override
	public void close() throws Exception {
		super.close();
		functionsClosed = true;
		FunctionUtils.closeFunction(userFunction);
	}
......

......
//生成上下文之后，接下来就是把上下文交给 SourceFunction 去执行,调用用户重写的run方法开始正式运行
			userFunction.run(ctx);

			// if we get here, then the user function either exited after being done (finite source)
			// or the function was canceled or stopped. For the finite source case, we should emit
			// a final watermark that indicates that we reached the end of event-time
			if (!isCanceledOrStopped()) {
				ctx.emitWatermark(Watermark.MAX_WATERMARK);
			}
......

//执行我们自己重写的 cancel 方法
public void cancel() {
		// important: marking the source as stopped has to happen before the function is stopped.
		// the flag that tracks this status is volatile, so the memory model also guarantees
		// the happens-before relationship
		markCanceledOrStopped();
		userFunction.cancel();

		// the context may not be initialized if the source was never running.
		if (ctx != null) {
			ctx.close();
		}
	}
......