Apache kafka connect-自定义websocket连接器

Question

目标是：开发一个自定义Kafka连接器，该连接器以循环方式从websocket读取消息。我试着给你们举一个我所认识到的例子：

我创建了一个接口IWebsocketClientEndpoint

    public interface IWebsocketClientEndpoint {
    IWebsocketClientEndpoint Connect() ;
    void Disconnect() throws IOException;
    IWebsocketClientEndpoint addMessageHandler(IMessageHandler msgHandler);
    void SendMessage(String message) throws Exception;
    void SendMessage(ByteBuffer message) throws Exception;
    void SendMessage(Object message) throws Exception;
    boolean isOpen();
    void Dispose()throws IOException;
}

以及实现上述接口的类：

@ClientEndpoint
public class WebsocketClientEndpoint implements IWebsocketClientEndpoint {

    private WebSocketContainer _container;
    private Session _userSession = null;
    private IMessageHandler _messageHandler;
    private URI _endpointURI;

    private WebsocketClientEndpoint(URI endpointURI) {
        try {
            _endpointURI = endpointURI;
            _container = ContainerProvider.getWebSocketContainer();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    private WebsocketClientEndpoint(URI endpointURI, int bufferSize) {
        try {
            _endpointURI = endpointURI;
            _container = ContainerProvider.getWebSocketContainer();
            _container.setDefaultMaxBinaryMessageBufferSize(bufferSize);
            _container.setDefaultMaxTextMessageBufferSize(bufferSize);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

  
    public static IWebsocketClientEndpoint Create(URI endpointURI){
        return new WebsocketClientEndpoint(endpointURI);
    }

    public static IWebsocketClientEndpoint Create(URI endpointURI,int bufferSize){
        return new WebsocketClientEndpoint(endpointURI,bufferSize);
    }

    public IWebsocketClientEndpoint Connect()  {
        try {
                      _container.connectToServer(this, _endpointURI);
            return this;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    @OnOpen
    public void onOpen(Session userSession) {
        this._userSession = userSession;

       if (this._messageHandler != null) {
            this._messageHandler.handleOpen("Web socket "+ _endpointURI +" opened");}
    }


    @OnClose
    public void onClose(Session userSession, CloseReason reason) {

        this._userSession = null;

        if (this._messageHandler != null) {
            this._messageHandler.handleClose("Web socket "+ _endpointURI +" closed. Reason: " + reason.getReasonPhrase());}
    }

    public void Disconnect() throws IOException {
        CloseReason reason  = new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE,"Web socket closed by user");

        this._userSession.close(reason);
        this._userSession = null;
        //close notification to caller
        if (this._messageHandler != null) {
            this._messageHandler.handleClose("Web socket "+ _endpointURI +" closed. Reason: " + reason.getReasonPhrase());}
    }

    @Override
    public IWebsocketClientEndpoint addMessageHandler(IMessageHandler msgHandler) {
        this._messageHandler = msgHandler;
        return  this;
    }

    @OnMessage
    public void onMessage(String message) {
        if (this._messageHandler != null) {
            this._messageHandler.handleMessage(message);
        }
    }

    @OnMessage
    public void onMessage(ByteBuffer bytes) {

        if (this._messageHandler != null) {
            this._messageHandler.handleMessage(bytes);
        }
    }

    public void SendMessage(String message) throws Exception {
        try{
           this._userSession.getAsyncRemote().sendText(message);
        }catch (Exception ex){
            throw ex;
        }
    }

    public void SendMessage(ByteBuffer message) throws Exception {
        try{
            this._userSession.getAsyncRemote().sendBinary(message);
        }catch (Exception ex){
            throw ex;
        }
    }


    public void SendMessage(Object message) throws Exception {
            this._userSession.getAsyncRemote().sendObject(message);
        }catch (Exception ex){
            throw ex;
        }
    }

    @Override
    public boolean isOpen() {
        if (this._userSession != null){
            return  this._userSession.isOpen();
        }
        return false;
    }
}

WebsocketClientEndpoint类专用于创建websocket并管理连接、断开连接、发送和接收消息。

目标是：如何在Kafka连接结构中调整我的websocket结构？我可以将从套接字接收到的消息（“public void handleMessage（String s））排入ConcurrentLinkedQueue，然后在kafka连接循环方法中，将其解列。但这是最好的解决方案吗？

下面，我的Kafka自定义连接器的实现

My kafka Connector

public class MySourceTask extends SourceTask {
IWebsocketClientEndpoint _clientEndPoint;

    @Override
    public void start(Map<String, String> props) {

                _clientEndPoint = WebsocketClientEndpoint
                        .Create(new URI(socket))
                        .Connect();

            _clientEndPoint.addMessageHandler(new IMessageHandler() {
                @Override
                public void handleMessage(String s) {

                }

                @Override
                public void handleMessage(ByteBuffer byteBuffer) {

                }

                @Override
                public void handleClose(String s) {

                }

                @Override
                public void handleOpen(String s) {

                }
            });
     
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        return null;
    }

    @Override
    public void stop() {
            _clientEndPoint.Dispose();
    }
}

提前感谢所有人

Answer 1

我建议将接口添加到类中

extends SourceTask implements IMessageHandler

然后呢

_clientEndPoint.addMessageHandler(this);

当您实现handleMessage时，将这些字符串添加到一些队列中。在poll方法中，您可以从该队列中弹出数据来创建要返回的SourceRecord对象。

在停止内，调用此。关闭并清理其他资源。