用Jetty 9.1运行Java WebSockets微服务

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介:   Jetty 9.1的发布将Java WebSockets (JSR-356) 带入了非Java EE环境,从而开启了微服务时代。我们可以将Jetty的容器包含在java应用程序中(注意,不是Java代码运行在容器中,而是相反),这种微服务轻量概念开始得到提倡推广,为模块化开启新的探索方向。

  Jetty 9.1的发布将Java WebSockets (JSR-356) 带入了非Java EE环境,从而开启了微服务时代。我们可以将Jetty的容器包含在java应用程序中(注意,不是Java代码运行在容器中,而是相反),这种微服务轻量概念开始得到提倡推广,为模块化开启新的探索方向。

  该案例目标是要建设一个从客户端程序接受消息并广播到当前连接的所有其他客户端WebSocket服务器。假设有一个消息模型:

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

客户端代码:

@ClientEndpoint
public class BroadcastClientEndpoint {
    private static final Logger log = Logger.getLogger( 
        BroadcastClientEndpoint.class.getName() );

    @OnOpen
    public void onOpen( final Session session ) throws IOException, EncodeException  {
        session.getBasicRemote().sendObject( new Message( "Client", "Hello!" ) );
    }

    @OnMessage
    public void onMessage( final Message message ) {
        log.info( String.format( "Received message '%s' from '%s'",
            message.getMessage(), message.getUsername() ) );
    }
}

@OnOpen 是当客户端连接到服务器开始调用,@OnMessage是每次服务器向客户端发送消息时调用。

消息传递使用Json,这里使用JSR-353规范的Json类将对象进行序列化。我们需要在Message里面加上一下Json反序列化,也就是将Json转为Message对象:

public class Message {
    public static class MessageDecoder implements Decoder.Text< Message > {
        private JsonReaderFactory factory = Json.createReaderFactory( Collections.< String, Object >emptyMap() );

        @Override
        public void init( final EndpointConfig config ) {
        }

        @Override
        public 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;
        }

        @Override
        public boolean willDecode( final String str ) {
            return true;
        }

        @Override
        public void destroy() {
        }
    }
}

我们需要告诉客户端,我们有一个Json编码器和解码器,在BroadcastClientEndpoint类上加入:

@ClientEndpoint( encoders = { MessageEncoder.class }, decoders = { MessageDecoder.class } )
public class BroadcastClientEndpoint {
}

下面是调用运行代码:

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();
    }
}

这是连接URL ws://localhost:8080/broadcast,随机挑选一些客户端名称(从UUID),每1秒的延迟产生10条信息,(只是为了确保我们有时间去接收他们都回来了)。

下面是服务器端的代码:

@ServerEndpoint( 
    value = "/broadcast", 
    encoders = { MessageEncoder.class }, 
    decoders = { MessageDecoder.class } 

public class BroadcastServerEndpoint {
    private static final Set< Session > sessions = 
        Collections.synchronizedSet( new HashSet< Session >() );

    @OnOpen
    public void onOpen( final Session session ) {
        sessions.add( session );
    }

    @OnClose
    public void onClose( final Session session ) {
        sessions.remove( session );
    }

    @OnMessage
    public void onMessage( final Message message, final Session client ) 
            throws IOException, EncodeException {
        for( final Session session: sessions ) {
            session.getBasicRemote().sendObject( message );
        }
    }
}

为了使这个服务器端点能够运行,我们将其注册入Jetty服务器,Jetty9.能够在嵌入下运行:

public class ServerStarter  {
    public static void main( String[] args ) throws Exception {
        Server server = new Server( 8080 );

        // Create the 'root' Spring application context
        final 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的AppConfig 能够帮助我们做到这点:

@Configuration
public class AppConfig  {
    @Inject private WebApplicationContext context;
    private ServerContainer container;

    public class SpringServerEndpointConfigurator extends ServerEndpointConfig.Configurator {
        @Override
        public < T > T getEndpointInstance( Class< T > endpointClass ) 
                throws InstantiationException {
            return context.getAutowireCapableBeanFactory().createBean( endpointClass );   
        }
    }

    @Bean
    public ServerEndpointConfig.Configurator configurator() {
        return new SpringServerEndpointConfigurator();
    }

    @PostConstruct
    public 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 )  
            ) {
                @Override
                public Configurator getConfigurator() {
                    return configurator();
                }
            }
        );
    }  
}

容器通过调用构造函数将创建container,然后每一次新的客户端连接创建一个服务器端点的新实例。

我们检索的WebSockets容器的方法是Jetty专用规范:查询来自名为“javax.websocket.server.ServerContainer”上下文的属性。

最后运行:

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

输出结果部分:

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

该项目源码下载: GitHub

目录
相关文章
|
2月前
|
监控 数据可视化 关系型数据库
微服务架构+Java+Spring Cloud +UniApp +MySql智慧工地系统源码
项目管理:项目名称、施工单位名称、项目地址、项目地址、总造价、总面积、施工准可证、开工日期、计划竣工日期、项目状态等。
304 6
|
2月前
|
人工智能 监控 安全
java基于微服务架构的智慧工地监管平台源码带APP
劳务管理: 工种管理、分包商管理、信息采集、班组管理、花名册、零工采集、 现场统计、考勤管理、考勤明细、工资管理、零工签证
286 4
|
3月前
|
Java 编译器
运行Java
本篇介绍了java程序的运行原理,包括javac和java两个命令的作用。javac :负责编译,当执行 javac 时,会启动 java 的编译器程序。对指定扩展名的 .java 文件进行编译。生成了jvm可以识别的字节码文件。java: 用来执行class字节码文件(即运行java程序)。
56 1
运行Java
|
2天前
|
负载均衡 Java 开发者
细解微服务架构实践:如何使用Spring Cloud进行Java微服务治理
【4月更文挑战第17天】Spring Cloud是Java微服务治理的首选框架,整合了Eureka(服务发现)、Ribbon(客户端负载均衡)、Hystrix(熔断器)、Zuul(API网关)和Config Server(配置中心)。通过Eureka实现服务注册与发现,Ribbon提供负载均衡,Hystrix实现熔断保护,Zuul作为API网关,Config Server集中管理配置。理解并运用Spring Cloud进行微服务治理是现代Java开发者的关键技能。
|
20天前
|
Java Maven Spring
SpringBoot运行出现 Lookup method resolution failed; nested exception is java.lang.IllegalStateException
SpringBoot运行出现 Lookup method resolution failed; nested exception is java.lang.IllegalStateException
26 0
|
22天前
|
Java Maven
运行maven项目出现Error:java: JDK isn‘t specified for module ‘XXX‘
运行maven项目出现Error:java: JDK isn‘t specified for module ‘XXX‘
11 0
|
1月前
|
传感器 人工智能 监控
智慧工地云信息平台源码(微服务+java+springcloud+uniapp+mysql)
智慧工地云信息平台源码(微服务+java+springcloud+uniapp+mysql)
30 0
|
1月前
|
设计模式 缓存 Java
Java新时代:微服务架构下的性能优化实践
【2月更文挑战第12天】 在当今快速发展的软件工程领域,微服务架构因其灵活性和可扩展性而成为主流。随着应用程序变得日益复杂,性能优化成为了开发者不可回避的挑战。本文将探讨在Java环境下,利用微服务架构进行性能优化的策略和实践。我们将从微服务的基本概念出发,深入分析如何通过设计模式、数据库优化、缓存机制以及并发处理等手段,有效提升Java应用的性能。此外,本文还将分享一些实际案例,以帮助读者更好地理解和应用这些优化技术。
|
2月前
|
存储 Java Docker
深入浅出:使用Docker容器化改进Java微服务架构
在快速演进的软件开发领域,微服务架构已成为构建可扩展、灵活且独立部署的应用程序的首选模式。然而,随着服务数量的增加,管理与部署这些服务的复杂性也随之上升。本文将探讨如何利用Docker容器技术,简化并优化Java微服务架构的开发、部署和运维流程。通过具体实例,我们将展示容器化如何为微服务架构带来的高效性、一致性和可伸缩性,从而使开发团队能够更加专注于业务逻辑的实现,而非环境的搭建与维护。
19 1
|
3月前
|
Dubbo Java 应用服务中间件
已跪,Java全能笔记爆火,分布式/开源框架/微服务/性能调优全有
程序员,立之根本还是技术,一个程序员的好坏,虽然不能完全用技术强弱来判断,但是技术水平一定是基础,技术差的程序员只能CRUD,技术不深的程序员也成不了架构师。程序员对于技术的掌握,除了从了解-熟悉-熟练-精通的过程以外,还应该从基础出发,到进阶,到源码,到实战。所以,程序员想要成功,首先要成就自己。