Jetty 9.1上的Java WebSockets（JSR-356）

最终发布了Jetty 9.1 ，将Java WebSockets（JSR-356）引入了非EE环境。这真是个好消息，今天的帖子将介绍如何将这个出色的新API与Spring Framework一起使用。

JSR-356定义了基于注释的简洁模型，以允许现代Java Web应用程序使用WebSockets API轻松创建双向通信通道。它不仅涵盖服务器端，还涵盖客户端端，这使得该API真正易于在任何地方使用。

让我们开始吧！我们的目标是构建一个WebSockets服务器，该服务器接受来自客户端的消息并将其广播到当前连接的所有其他客户端。首先，让我们定义消息格式，作为此简单的Message类，它将交换服务器和客户端。我们可以将自己限制为String之类的东西，但我想向您介绍另一个新API的功能– 用于JSON处理的Java API（JSR-353）。

package com.example.services;public class Message {private String username;private String message;public Message() {}public Message( final String username, final String message ) {this.username = username;this.message = message;}public String getMessage() {return message;}public String getUsername() {return username;}public void setMessage( final String message ) {this.message = message;}public void setUsername( final String username ) {this.username = username;}
}

为了分隔与服务器和客户端有关的声明， JSR-356定义了两个基本注释：分别为@ServerEndpoint和@ClientEndpoit 。我们的客户端端点，我们将其称为BroadcastClientEndpoint ，将仅侦听服务器发送的消息：

package com.example.services;import java.io.IOException;
import java.util.logging.Logger;import javax.websocket.ClientEndpoint;
import javax.websocket.EncodeException;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;@ClientEndpoint
public class BroadcastClientEndpoint {private static final Logger log = Logger.getLogger( BroadcastClientEndpoint.class.getName() );@OnOpenpublic void onOpen( final Session session ) throws IOException, EncodeException  {session.getBasicRemote().sendObject( new Message( "Client", "Hello!" ) );}@OnMessagepublic void onMessage( final Message message ) {log.info( String.format( "Received message '%s' from '%s'",message.getMessage(), message.getUsername() ) );}
}

就是这样！很干净，的代码不言自明的片：当客户端得到了连接到服务器和@ 的onMessage被称为每次服务器@OnOpen被称为发送消息给客户端。是的，它非常简单，但是有一个警告： JSR-356实现可以处理任何简单对象，但不能处理诸如Message is之类的复杂对象。为了解决这个问题， JSR-356引入了编码器和解码器的概念。

我们都喜欢JSON ，那么为什么不定义自己的JSON编码器和解码器呢？ JSON处理的Java API（JSR-353）可以为我们完成这项简单的任务。要创建编码器，只需实现Encoder.Text <Message>并使用JsonObjectBuilder将对象基本序列化为某个字符串，在我们的情况下为JSON字符串。

package com.example.services;import javax.json.Json;
import javax.json.JsonReaderFactory;
import javax.websocket.EncodeException;
import javax.websocket.Encoder;
import javax.websocket.EndpointConfig;public class Message {public static class MessageEncoder implements Encoder.Text< Message > {@Overridepublic void init( final EndpointConfig config ) {}@Overridepublic String encode( final Message message ) throws EncodeException {return Json.createObjectBuilder().add( "username", message.getUsername() ).add( "message", message.getMessage() ).build().toString();}@Overridepublic void destroy() {}}
}

对于解码器部分，一切看起来都非常相似，我们必须实现Decoder.Text <Message>并使用JsonReader从字符串反序列化我们的对象。

package com.example.services;import javax.json.JsonObject;
import javax.json.JsonReader;
import javax.json.JsonReaderFactory;
import javax.websocket.DecodeException;
import javax.websocket.Decoder;public class Message {public static class MessageDecoder implements Decoder.Text< Message > {private JsonReaderFactory factory = Json.createReaderFactory( Collections.< String, Object >emptyMap() );@Overridepublic void init( final EndpointConfig config ) {}@Overridepublic Message decode( final String str ) throws DecodeException {final Message message = new Message();try( final JsonReader reader = factory.createReader( new StringReader( str ) ) ) {final JsonObject json = reader.readObject();message.setUsername( json.getString( "username" ) );message.setMessage( json.getString( "message" ) );}return message;}@Overridepublic boolean willDecode( final String str ) {return true;}@Overridepublic void destroy() {}}
}

最后，我们需要告诉客户端（和服务器，它们共享相同的解码器和编码器），我们拥有用于消息的编码器和解码器。最简单的方法就是将它们声明为@ServerEndpoint和@ClientEndpoit批注的一部分。

import com.example.services.Message.MessageDecoder;
import com.example.services.Message.MessageEncoder;@ClientEndpoint( encoders = { MessageEncoder.class }, decoders = { MessageDecoder.class } )
public class BroadcastClientEndpoint {
}

为了使客户的示例更完整，我们需要某种方式使用BroadcastClientEndpoint连接到服务器并基本上交换消息。 ClientStarter类最终确定图片：

package com.example.ws;import java.net.URI;
import java.util.UUID;import javax.websocket.ContainerProvider;
import javax.websocket.Session;
import javax.websocket.WebSocketContainer;import org.eclipse.jetty.websocket.jsr356.ClientContainer;import com.example.services.BroadcastClientEndpoint;
import com.example.services.Message;public class ClientStarter {public static void main( final String[] args ) throws Exception {final String client = UUID.randomUUID().toString().substring( 0, 8 );final WebSocketContainer container = ContainerProvider.getWebSocketContainer();    final String uri = "ws://localhost:8080/broadcast";  try( Session session = container.connectToServer( BroadcastClientEndpoint.class, URI.create( uri ) ) ) {for( int i = 1; i <= 10; ++i ) {session.getBasicRemote().sendObject( new Message( client, "Message #" + i ) );Thread.sleep( 1000 );}}// Application doesn't exit if container's threads are still running( ( ClientContainer )container ).stop();}
}

只需几句注释，此代码即可完成操作：我们正在ws：// localhost：8080 / broadcast连接到WebSockets端点，随机选择一些客户端名称（来自UUID）并生成10条消息，每条消息都有1秒的延迟（请确保我们有时间将它们全部收回来）。

服务器部分看起来并没有很大不同，并且在这一点上无需任何其他注释就可以理解（服务器可能只是将接收到的每条消息广播给所有连接的客户端）。这里要提到的重要一点：每当新客户端连接到服务器端点时，都会创建服务器端点的新实例（这就是对等体集合是静态的），这是默认行为，可以轻松更改。

package com.example.services;import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;import javax.websocket.EncodeException;
import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;import com.example.services.Message.MessageDecoder;
import com.example.services.Message.MessageEncoder;@ServerEndpoint( value = "/broadcast", encoders = { MessageEncoder.class }, decoders = { MessageDecoder.class } 
) 
public class BroadcastServerEndpoint {private static final Set< Session > sessions = Collections.synchronizedSet( new HashSet< Session >() ); @OnOpenpublic void onOpen( final Session session ) {sessions.add( session );}@OnClosepublic void onClose( final Session session ) {sessions.remove( session );}@OnMessagepublic void onMessage( final Message message, final Session client ) throws IOException, EncodeException {for( final Session session: sessions ) {session.getBasicRemote().sendObject( message );}}
}

为了使该端点可用于连接，我们应该启动WebSockets容器并在其中注册该端点。与往常一样， Jetty 9.1可以轻松地以嵌入式模式运行：

package com.example.ws;import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.servlet.DefaultServlet;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.websocket.jsr356.server.deploy.WebSocketServerContainerInitializer;
import org.springframework.web.context.ContextLoaderListener;
import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;import com.example.config.AppConfig;public class ServerStarter  {public static void main( String[] args ) throws Exception {Server server = new Server( 8080 );// Create the 'root' Spring application contextfinal ServletHolder servletHolder = new ServletHolder( new DefaultServlet() );final ServletContextHandler context = new ServletContextHandler();context.setContextPath( "/" );context.addServlet( servletHolder, "/*" );context.addEventListener( new ContextLoaderListener() );   context.setInitParameter( "contextClass", AnnotationConfigWebApplicationContext.class.getName() );context.setInitParameter( "contextConfigLocation", AppConfig.class.getName() );server.setHandler( context );WebSocketServerContainerInitializer.configureContext( context );        server.start();server.join(); }
}

上面片段中最重要的部分是WebSocketServerContainerInitializer.configureContext ：它实际上是创建WebSockets容器的实例。因为我们还没有添加任何端点，所以容器基本上位于此处，什么也不做。 Spring Framework和AppConfig配置类将为我们完成最后的连接。

package com.example.config;import javax.annotation.PostConstruct;
import javax.inject.Inject;
import javax.websocket.DeploymentException;
import javax.websocket.server.ServerContainer;
import javax.websocket.server.ServerEndpoint;
import javax.websocket.server.ServerEndpointConfig;import org.eclipse.jetty.websocket.jsr356.server.AnnotatedServerEndpointConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.context.WebApplicationContext;import com.example.services.BroadcastServerEndpoint;@Configuration
public class AppConfig  {@Inject private WebApplicationContext context;private ServerContainer container;public class SpringServerEndpointConfigurator extends ServerEndpointConfig.Configurator {@Overridepublic < T > T getEndpointInstance( Class< T > endpointClass ) throws InstantiationException {return context.getAutowireCapableBeanFactory().createBean( endpointClass );   }}@Beanpublic ServerEndpointConfig.Configurator configurator() {return new SpringServerEndpointConfigurator();}@PostConstructpublic void init() throws DeploymentException {container = ( ServerContainer )context.getServletContext().getAttribute( javax.websocket.server.ServerContainer.class.getName() );container.addEndpoint( new AnnotatedServerEndpointConfig( BroadcastServerEndpoint.class, BroadcastServerEndpoint.class.getAnnotation( ServerEndpoint.class )  ) {@Overridepublic Configurator getConfigurator() {return configurator();}});}  
}

如前所述，默认情况下，容器会在每次新客户端连接时创建服务器端点的新实例，并通过调用构造函数来实现，在本例中为BroadcastServerEndpoint.class.newInstance（） 。这可能是理想的行为，但是由于我们使用的是Spring Framework和依赖项注入，因此此类新对象基本上是不受管的bean。由于JSR-356的精心设计（在我看来），通过实现ServerEndpointConfig.Configurator可以很容易地提供您自己的创建端点实例的方式。 SpringServerEndpointConfigurator就是这种实现方式的一个示例：每次请求新的终结点实例时，它都会创建一个新的托管Bean（如果您想要单个实例，则可以在AppConfig中将终结点的一个实例创建为Bean并一直返回）。

我们检索WebSockets容器的方式特定于Jetty ：从名称为“ javax.websocket.server.ServerContainer”的上下文属性中获取（将来可能会更改）。容器到达后，我们通过提供我们自己的ServerEndpointConfig （基于Jetty已经提供的AnnotatedServerEndpointConfig ）来添加新的（托管！）端点。

要构建和运行我们的服务器和客户端，我们只需要这样做：

mvn clean package
java -jar target\jetty-web-sockets-jsr356-0.0.1-SNAPSHOT-server.jar // run server
java -jar target/jetty-web-sockets-jsr356-0.0.1-SNAPSHOT-client.jar // run yet another client

例如，通过运行服务器和几个客户端（我运行了其中的四个，即“ 392f68ef ”，“ 8e3a869d ”，“ ca3a06d0 ”，“ 6cb82119 ”），您可能会在控制台的输出中看到每个客户端都接收到所有消息来自所有其他客户端（包括其自身的消息）：

Nov 29, 2013 9:21:29 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Hello!' from 'Client'
Nov 29, 2013 9:21:29 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #1' from '392f68ef'
Nov 29, 2013 9:21:29 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #2' from '8e3a869d'
Nov 29, 2013 9:21:29 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #7' from 'ca3a06d0'
Nov 29, 2013 9:21:30 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #4' from '6cb82119'
Nov 29, 2013 9:21:30 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #2' from '392f68ef'
Nov 29, 2013 9:21:30 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #3' from '8e3a869d'
Nov 29, 2013 9:21:30 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #8' from 'ca3a06d0'
Nov 29, 2013 9:21:31 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #5' from '6cb82119'
Nov 29, 2013 9:21:31 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #3' from '392f68ef'
Nov 29, 2013 9:21:31 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #4' from '8e3a869d'
Nov 29, 2013 9:21:31 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #9' from 'ca3a06d0'
Nov 29, 2013 9:21:32 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #6' from '6cb82119'
Nov 29, 2013 9:21:32 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #4' from '392f68ef'
Nov 29, 2013 9:21:32 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #5' from '8e3a869d'
Nov 29, 2013 9:21:32 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #10' from 'ca3a06d0'
Nov 29, 2013 9:21:33 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #7' from '6cb82119'
Nov 29, 2013 9:21:33 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #5' from '392f68ef'
Nov 29, 2013 9:21:33 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #6' from '8e3a869d'
Nov 29, 2013 9:21:34 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #8' from '6cb82119'
Nov 29, 2013 9:21:34 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #6' from '392f68ef'
Nov 29, 2013 9:21:34 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #7' from '8e3a869d'
Nov 29, 2013 9:21:35 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #9' from '6cb82119'
Nov 29, 2013 9:21:35 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #7' from '392f68ef'
Nov 29, 2013 9:21:35 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #8' from '8e3a869d'
Nov 29, 2013 9:21:36 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #10' from '6cb82119'
Nov 29, 2013 9:21:36 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #8' from '392f68ef'
Nov 29, 2013 9:21:36 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #9' from '8e3a869d'
Nov 29, 2013 9:21:37 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #9' from '392f68ef'
Nov 29, 2013 9:21:37 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #10' from '8e3a869d'
Nov 29, 2013 9:21:38 PM com.example.services.BroadcastClientEndpoint onMessage
INFO: Received message 'Message #10' from '392f68ef'
2013-11-29 21:21:39.260:INFO:oejwc.WebSocketClient:main: Stopped org.eclipse.jetty.websocket.client.WebSocketClient@3af5f6dc

太棒了！我希望这篇介绍性博客文章能够显示在Java中使用现代Web通信协议变得多么容易，这要归功于Java WebSockets（JSR-356），用于JSON处理的Java API（JSR-353）以及诸如Jetty 9.1之类的出色项目！