本篇文章给大家谈谈php实现websocket实时消息推送，以及phpwebsocket推送的知识点，同时本文还将给你拓展javaWebSocket实现聊天消息推送功能、Java通过Netty，实现W
本篇文章给大家谈谈php实现websocket实时消息推送，以及php websocket推送的知识点，同时本文还将给你拓展java WebSocket实现聊天消息推送功能、Java通过Netty，实现Websocket消息推送简单几步搞定、JS实现websocket实时消息提示、JS实现websocket长轮询实时消息提示的效果等相关知识，希望对各位有所帮助，不要忘了收藏本站喔。
本文目录一览：- php实现websocket实时消息推送（php websocket推送）
- java WebSocket实现聊天消息推送功能
- Java通过Netty，实现Websocket消息推送简单几步搞定
- JS实现websocket实时消息提示
- JS实现websocket长轮询实时消息提示的效果
php实现websocket实时消息推送（php websocket推送）
PHP实现websocket实时消息推送，供大家参考，具体内容如下
SocketService.PHP
{
private $address = '0.0.0.0';
private $port = 8083;
private $_sockets;
public function __construct($address = '',$port='')
{
if(!empty($address)){
$this->address = $address;
}
if(!empty($port)) {
$this->port = $port;
}
}
public function service(){
//获取tcp协议号码。
$tcp = getprotobyname("tcp");
$sock = socket_create(AF_INET,SOCK_STREAM,$tcp);
socket_set_option($sock,SOL_SOCKET,SO_REUSEADDR,1);
if($sock < 0)
{
throw new Exception("failed to create socket: ".socket_strerror($sock)."\n");
}
socket_bind($sock,$this->address,$this->port);
socket_listen($sock,$this->port);
echo "listen on $this->address $this->port ... \n";
$this->_sockets = $sock;
}
public function run(){
$this->service();
$clients[] = $this->_sockets;
while (true){
$changes = $clients;
$write = NULL;
$except = NULL;
socket_select($changes,$write,$except,NULL);
foreach ($changes as $key => $_sock){
if($this->_sockets == $_sock){ //判断是不是新接入的socket
if(($newClient = socket_accept($_sock)) === false){
die('Failed to accept socket: '.socket_strerror($_sock)."\n");
}
$line = trim(socket_read($newClient,1024));
$this->handshaking($newClient,$line);
//获取client ip
socket_getpeername ($newClient,$ip);
$clients[$ip] = $newClient;
echo "Client ip:{$ip} \n";
echo "Client msg:{$line} \n";
} else {
socket_recv($_sock,$buffer,2048,0);
$msg = $this->message($buffer);
//在这里业务代码
echo "{$key} clinet msg:",$msg,"\n";
fwrite(STDOUT,'Please input a argument:');
$response = trim(fgets(STDIN));
$this->send($_sock,$response);
echo "{$key} response to Client:".$response,"\n";
}
}
}
}
/**
- 握手处理
- @param $newClient socket
- @return int 接收到的信息
*/
public function handshaking($newClient,$line){
$headers = array();$li<a href="https://www.jb51.cc/tag/nes/" target="_blank" class="keywords">nes</a> = preg_split("/\r\n/",$line);foreach($li<a href="https://www.jb51.cc/tag/nes/" target="_blank">nes</a> as $line){$line = chop($line);if(preg_match('/\A(\S+): (.*)\z/',$line,$matches)){$headers[$matches[1]] = $matches[2];}}$secKey = $headers['Sec-WebSocket-Key'];$secAccept = base64_encode(pack('H*',sha1($secKey . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')));$upgrade = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" ."Upgrade: websocket\r\n" ."Connection: Upgrade\r\n" ."WebSocket-Origin: $this->address\r\n" ."WebSocket-Location: ws://$this->address:$this->port/websocket/websocket\r\n"."Sec-WebSocket-Accept:$secAccept\r\n\r\n";return socket_write($newClient,$upgrade,strlen($upgrade));
}
/**
- 解析接收数据
- @param $buffer
- @return null|string
*/
public function message($buffer){
$len = $masks = $data = $decoded = null;
$len = ord($buffer[1]) & 127;
if ($len === 126) {
$masks = substr($buffer,4,4);
$data = substr($buffer,8);
} else if ($len === 127) {
$masks = substr($buffer,10,14);
} else {
$masks = substr($buffer,2,6);
}
for ($index = 0; $index < strlen($data); $index++) {
$decoded .= $data[$index] ^ $masks[$index % 4];
}
return $decoded;
}
/**
- 发送数据
- @param $newClinet 新接入的socket
- @param $msg 要发送的数据
- @return int|string
*/
public function send($newClinet,$msg){
$msg = $this->frame($msg);
socket_write($newClinet,strlen($msg));
}
public function frame($s) {
$a = str_split($s,125);
if (count($a) == 1) {
return "\x81" . chr(strlen($a[0])) . $a[0];
}
$ns = "";
foreach ($a as $o) {
$ns .= "\x81" . chr(strlen($o)) . $o;
}
return $ns;
}
/**
- 关闭socket
*/
public function close(){
return socket_close($this->_sockets);
}
}
$sock = new SocketService();
$sock->run();
web.html
<div>
<prehttps://www.jb51.cc/tag/rush/" target="_blank">rush:xhtml;">
<!doctype html>
<html lang="en">
java WebSocket实现聊天消息推送功能
这篇文章主要为大家详细介绍了java WebSocket实现聊天消息推送功能，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
本文实例为大家分享了java WebSocket实现聊天消息推送功能的具体代码，供大家参考，具体内容如下
环境：
JDK.1.7.0_51
apache-tomcat-7.0.53
java jar包：tomcat-coyote.jar、tomcat-juli.jar、websocket-api.jar
Chatannotation消息发送类：
import java.io.IOException; import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import javax.websocket.OnClose; import javax.websocket.OnError; import javax.websocket.OnMessage; import javax.websocket.Onopen; import javax.websocket.Session; import javax.websocket.server.ServerEndpoint; import org.apache.juli.logging.Log; import org.apache.juli.logging.LogFactory; import com.util.HTMLFilter; /** * WebSocket 消息推送服务类 * @author 胡汉三 * * 2014-11-18 下午7:53:13 */ @ServerEndpoint(value = "/websocket/chat") public class Chatannotation { private static final Log log = LogFactory.getLog(Chatannotation.class); private static final String GUEST_PREFIX = "Guest"; private static final AtomicInteger connectionIds = new AtomicInteger(0); private static final Map connections = new HashMap(); private final String nickname; private Session session; public Chatannotation() { nickname = GUEST_PREFIX + connectionIds.getAndIncrement(); } @Onopen public void start(Session session) { this.session = session; connections.put(nickname, this); String message = String.format("* %s %s", nickname, "has joined."); broadcast(message); } @OnClose public void end() { connections.remove(this); String message = String.format("* %s %s", nickname, "has disconnected."); broadcast(message); } /** * 消息发送触发方法 * @param message */ @OnMessage public void incoming(String message) { // Never trust the client String filteredMessage = String.format("%s: %s", nickname, HTMLFilter.filter(message.toString())); broadcast(filteredMessage); } @OnError public void onError(Throwable t) throws Throwable { log.error("Chat Error: " + t.toString(), t); } /** * 消息发送方法 * @param msg */ private static void broadcast(String msg) { if(msg.indexOf("Guest0")!=-1){ sendUser(msg); } else{ sendAll(msg); } } /** * 向所有用户发送 * @param msg */ public static void sendAll(String msg){ for (String key : connections.keySet()) { Chatannotation client = null ; try { client = (Chatannotation) connections.get(key); synchronized (client) { client.session.getBasicRemote().sendText(msg); } } catch (IOException e) { log.debug("Chat Error: Failed to send message to client", e); connections.remove(client); try { client.session.close(); } catch (IOException e1) { // Ignore } String message = String.format("* %s %s", client.nickname, "has been disconnected."); broadcast(message); } } } /** * 向指定用户发送消息 * @param msg */ public static void sendUser(String msg){ Chatannotation c = (Chatannotation)connections.get("Guest0"); try { c.session.getBasicRemote().sendText(msg); } catch (IOException e) { log.debug("Chat Error: Failed to send message to client", e); connections.remove(c); try { c.session.close(); } catch (IOException e1) { // Ignore } String message = String.format("* %s %s", c.nickname, "has been disconnected."); broadcast(message); } } }
HTMLFilter工具类：
/** * HTML 工具类 * * @author 胡汉三 */ public final class HTMLFilter { public static String filter(String message) { if (message == null) return (null); char content[] = new char[message.length()]; message.getChars(0, message.length(), content, 0); StringBuilder result = new StringBuilder(content.length + 50); for (int i = 0; i ': result.append(">"); break; case '&': result.append("&"); break; case '"': result.append("""); break; default: result.append(content[i]); } } return (result.toString()); } }
页面：
测试
Seems your browser doesn't support Javascript! Websockets rely on Javascript being enabled. Please enable Javascript and reload this page!
可指定发送给某个用户，也可全部发送，详情见Chatannotation类的broadcast方法。
程序发布时记得删除tomcat-coyote.jar、tomcat-juli.jar、websocket-api.jar这三个jar包在启动Tomcat。
程序截图，Guest0用户发送信息的信息，在后台进行了判断只发送给自己：
Guest1：
Guest2：
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持小编。
Java通过Netty，实现Websocket消息推送简单几步搞定
前言
曾几何时，不知道大家有没有在项目里遇到过需要服务端给客户端推送消息的需求，是否曾经苦恼过、纠结过，我们知道要想实现这样的需求肯定离不开websocket长连接方式，那么到底是该选原生的websocket还是更加高级的netty框架呢？
在此我极力推荐netty，因为一款好的框架一般都是在原生的基础上进行包装成更好、更方便、更实用的东西，很多我们需要自己考虑的问题都基本可以不用去考虑，不过此文不会去讲netty有多么的高深莫测，因为这些概念性的东西随处可见，而是通过实战来达到推送消息的目的。
实战
一、逻辑架构图
从图中可以看出本次实战的基本流程是客户端A请求服务端核心模块，核心模块生产一条消息到消息队列，然后服务端消息模块消费消息，消费完之后就将消息推送给客户端B，流程很简单，没有太多技巧，唯一的巧妙之处就在消息模块这边的处理上，本文的重点也主要讲解消息模块这一块，主要包括netty server、netty client、channel的存储等等。
二、代码
1、添加依赖
<dependency>
<groupId>io.nettygroupId>
<artifactId>netty-allartifactId>
<version>4.1.6.Finalversion>
dependency>2、NettyServer类
@Service
public class NettyServer {
public void run(int port){
new Thread(){
public void run(){
runServer(port);
}
}.start();
}
private void runServer(int port){
Print.info("===============Message服务端启动===============");
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
b.childHandler(new ChannelInitializer
() {
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(
"codec-http",
new HttpServerCodec());
pipeline.addLast(
"aggregator",
new HttpObjectAggregator(
65536));
pipeline.addLast(
"handler",
new MyWebSocketServerHandler());
}
});
Channel ch = b.bind(port).sync().channel();
Print.info(
"Message服务器启动成功：" + ch.toString());
ch.closeFuture().sync();
}
catch (Exception e){
Print.error(
"Message服务运行异常：" + e.getMessage());
e.printStackTrace();
}
finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
Print.info(
"Message服务已关闭");
}
}
}
3、MyWebSocketServerHandler类
public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object>{
private static final String WEBSOCKET_PATH = "";
private WebSocketServerHandshaker handshaker;
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof FullHttpRequest){
//以http请求形式接入，但是走的是websocket
handleHttpRequest(ctx, (FullHttpRequest) msg);
}else if (msg instanceof WebSocketFrame){
//处理websocket客户端的消息
handleWebSocketFrame(ctx, (WebSocketFrame) msg);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
//要求Upgrade为websocket，过滤掉get/Post
if (!req.decoderResult().isSuccess()
|| (!"websocket".equals(req.headers().get("Upgrade")))) {
//若不是websocket方式，则创建BAD_REQUEST的req，返回给客户端
sendHttpResponse(ctx, req, new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
return;
}
WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
"ws://localhost:9502/websocket", null, false);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {
WebSocketServerHandshakerFactory
.sendUnsupportedVersionResponse(ctx.channel()); } else {
handshaker.handshake(ctx.channel(), req); }
}
private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
// Check for closing frame if (frame instanceof CloseWebSocketFrame) {
handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain()); return; }
if (frame instanceof PingWebSocketFrame) {
ctx.channel().write(new PongWebSocketFrame(frame.content().retain())); return; }
if (!(frame instanceof TextWebSocketFrame)) {
Print.error("数据帧类型不支持!"); throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName())); }
// Send the uppercase string back. String request = ((TextWebSocketFrame) frame).text(); Print.info("Netty服务器接收到的信息: " + request); if (request.equals(Const.HEARTBEAT)){
ctx.channel().write(new TextWebSocketFrame(request)); return; }
JSONObject jsonData = JSONObject.parseObject(request); String eventType = jsonData.getString("event_type"); String apiToken = jsonData.getString("api_token"); if (Const.FRONT.equals(eventType)){
Print.info("front event"); ChannelSupervise.updateChannel(apiToken, ctx.channel()); }else if (Const.BEHIND.equals(eventType)){
Print.info("behind event"); Channel chan = ChannelSupervise.findChannel(apiToken); if (null == chan){
Print.error("目标用户不存在"); }else {
JSONObject jsonMsg = new JSONObject(); jsonMsg.put("type", jsonData.get("type")); jsonMsg.put("child_type", jsonData.get("child_type")); jsonMsg.put("title", jsonData.get("title")); jsonMsg.put("body", jsonData.get("body")); ChannelSupervise.sendToSimple(apiToken, new TextWebSocketFrame(jsonMsg.toString())); Print.info("向目标用户发送成功"); }
}else{
Print.error("event type error"); }
}
private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
// 返回应答给客户端 if (res.status().code() != 200) {
ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8); res.content().writeBytes(buf); buf.release(); }
ChannelFuture f = ctx.channel().writeAndFlush(res); // 如果是非Keep-Alive，关闭连接 if (!isKeepAlive(req) || res.status().code() != 200) {
f.addListener(ChannelFutureListener.CLOSE); }
}
@Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace(); ctx.close(); }
private static String getWebSocketLocation(FullHttpRequest req) {
return "ws://" + req.headers().get(HOST) + WEBSOCKET_PATH; }
/** * 接收客户端连接事件 */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception {
Print.info("客户端与服务端连接开启：" + ctx.channel()); ChannelSupervise.addChannel(null, ctx.channel()); }
/** * 接收客户端关闭事件 */ @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Print.info("客户端与服务端连接关闭：" + ctx.channel()); ChannelSupervise.removeChannel(ctx.channel()); }
}4、ChannelSupervise类
public class ChannelSupervise {
private static ChannelGroup GlobalGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
private static ConcurrentMap
ChannelMap =
new ConcurrentHashMap();
public static void addChannel(String apiToken, Channel channel){
GlobalGroup.
add(channel);
if (
null != apiToken) {
ChannelMap.put(apiToken, channel.id());
}
}
public static void updateChannel(String apiToken, Channel channel){
Channel chan = GlobalGroup.find(channel.id());
if (
null == chan){
addChannel(apiToken, channel);
}
else {
ChannelMap.put(apiToken, channel.id());
}
}
public static void removeChannel(Channel channel){
GlobalGroup.
remove(channel);
Collection
values = ChannelMap.values();
values.
remove(channel.id());
}
public static Channel findChannel(String apiToken){
ChannelId chanId = ChannelMap.
get(apiToken);
if (
null == chanId){
return
null;
}
return GlobalGroup.find(ChannelMap.
get(apiToken));
}
public static void sendToAll(TextWebSocketFrame tws){
GlobalGroup.writeAndFlush(tws);
}
public static void sendToSimple(String apiToken, TextWebSocketFrame tws){
GlobalGroup.find(ChannelMap.
get(apiToken)).writeAndFlush(tws);
}
}
5、NettyClient类
@Servicepublic class NettyClient {
private Channel channel;
public void run(String strUri){
new Thread(){
public void run(){
runClient(strUri);
}
}.start();
private void runClient(String strUri) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
URI uri = new URI(strUri);
String protocol = uri.getScheme();
if (!"ws".equals(protocol)) {
throw new IllegalArgumentException("Unsupported protocol: " + protocol);
}
HttpHeaders customHeaders = new DefaultHttpHeaders();
customHeaders.add("MyHeader", "MyValue");
// Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
// If you change it to V00, ping is not supported and remember to change
// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
final MyWebSocketClientHandler handler =
new MyWebSocketClientHandler(
WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, customHeaders)); b.group(group); b.channel(NioSocketChannel.class); b.handler(new ChannelInitializer
() {
@
Overpublic void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(
"http-codec",
new HttpClientCodec()); pipeline.addLast(
"aggregator",
new HttpObjectAggregator(
8192)); pipeline.addLast(
"ws-handler", handler); }
}); Print.info(
"===============Message客户端启动==============="); channel = b.connect(uri.getHost(), uri.getPort()).sync().channel(); handler.handshakeFuture().sync(); channel.closeFuture().sync(); }
catch (Exception e){
Print.error(e.getMessage()); }
finally {
group.shutdownGracefully(); }
}
6、MyWebSocketClientHandler类
public class MyWebSocketClientHandler extends SimpleChannelInboundHandler<Object> {
private final WebSocketClientHandshaker handshaker;
private ChannelPromise handshakeFuture;
public MyWebSocketClientHandler(WebSocketClientHandshaker handshaker) {
this.handshaker = handshaker;
}
public ChannelFuture handshakeFuture() {
return handshakeFuture;
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
handshakeFuture = ctx.newPromise();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
handshaker.handshake(ctx.channel());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Print.info("webSocket client disconnected!");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Channel ch = ctx.channel();
if (!handshaker.isHandshakeComplete()) {
handshaker.finishHandshake(ch, (FullHttpResponse) msg);
Print.info("websocket client connected!");
handshakeFuture.setSuccess();
return;
}
if (msg instanceof FullHttpResponse) {
FullHttpResponse response = (FullHttpResponse) msg;
throw new Exception("Unexpected FullHttpResponse (getStatus=" + response.getStatus() + ", content=" + response.content().toString(CharsetUtil.UTF_8) + '')'');
}
WebSocketFrame frame = (WebSocketFrame) msg;
if (frame instanceof TextWebSocketFrame) {
TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
Print.info("客户端收到消息: " + textFrame.text());
} else if (frame instanceof PongWebSocketFrame) {
Print.info("websocket client received pong");
} else if (frame instanceof CloseWebSocketFrame) {
Print.info("websocket client received closing");
ch.close();
}
}
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
if (!handshakeFuture.isDone()) {
handshakeFuture.setFailure(cause);
}
ctx.close();
}
}7、启动类
@SpringBootApplication
@Servicepublic
class MessageApplication {
@Autowired
private NettyServer server;
@Autowired
private NettyClient client;
public static void main(String[] args) {
SpringApplication.run(MessageApplication.class, args);
}
@PostConstruct
public void initMessage(){
server.run(9502);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
client.run("ws://localhost:" + 9502);
}8、客户端B测试页面
<html>
<head>
<meta charset="UTF-8">
<title>WebSocket Chattitle>
head>
<body>
<script type="text/javascript">
var socket;
if (!window.WebSocket) {
window.WebSocket = window.MozWebSocket;
}
if (window.WebSocket) {
socket = new WebSocket("ws://localhost:9502");
socket.onmessage = function(event) {
var ta = document.getElementById(''responseText'');
ta.value = ta.value + ''\n'' + event.data
};
socket.onopen = function(event) {
var ta = document.getElementById(''responseText'');
ta.value = "连接开启!";
};
socket.onclose = function(event) {
var ta = document.getElementById(''responseText'');
ta.value = ta.value + "连接被关闭";
};
} else {
alert("你的浏览器不支持 WebSocket！");
}
function send(message) {
if (!window.WebSocket) {
return;
}
if (socket.readyState == WebSocket.OPEN) {
socket.send(message);
} else {
alert("连接没有开启.");
}
}
script>
<form onsubmit="return false;">
<h3>WebSocket：h3>
<textarea id="responseText" style="width: 500px; height: 300px;">textarea>
<br>
<input type="text" name="message" style="width: 300px" value="1">
<input type="button" value="发送消息" onclick="send(this.form.message.value)">
<input type="button" onclick="javascript:document.getElementById(''responseText'').value=''''" value="清空聊天记录">
form>
<br>
body>
html>三、测试
1、先运行启动类，此时会先启动netty服务器，然后启动一个netty客户端，然后过30s模拟客户端A进行消息发送
2、打开测试页面，在底下的输入框输入：{"event_type":"front", "api_token":"11111"}，表示客户端B连接上netty服务器
测试结果如下：
消息模块：
客户端B：
四、结束语
本文只是抛砖引玉，主要启发有类似需求的朋友知道怎么去存储channel，进而怎么给指定客户推送消息，如果想要进行大型项目的高并发、可靠稳定地使用，还需进一步地改进。
作者：都市心声
来源：toutiao.com/i6794445371457143307
