服务的心跳机制与断线重连,Netty底层是怎么实现的?
作者:sprinkle_liz
www.jianshu.com/p/1a28e48edd92
提醒:本篇适合有一定
netty
基础的读者阅读
心跳机制
何为心跳
所谓心跳, 即在 TCP
长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP
连接的有效性.
注:心跳包还有另一个作用,经常被忽略,即:一个连接如果长时间不用,防火墙或者路由器就会断开该连接。
如何实现
核心Handler —— IdleStateHandler
在 Netty
中, 实现心跳机制的关键是 IdleStateHandler
, 那么这个 Handler
如何使用呢? 先看下它的构造器:
public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
}
这里解释下三个参数的含义:
readerIdleTimeSeconds: 读超时. 即当在指定的时间间隔内没有从
Channel
读取到数据时, 会触发一个READER_IDLE
的IdleStateEvent
事件.writerIdleTimeSeconds: 写超时. 即当在指定的时间间隔内没有数据写入到
Channel
时, 会触发一个WRITER_IDLE
的IdleStateEvent
事件.allIdleTimeSeconds: 读/写超时. 即当在指定的时间间隔内没有读或写操作时, 会触发一个
ALL_IDLE
的IdleStateEvent
事件.
注:这三个参数默认的时间单位是秒。若需要指定其他时间单位,可以使用另一个构造方法:
IdleStateHandler(boolean observeOutput, long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit)
在看下面的实现之前,建议先了解一下IdleStateHandler
的实现原理。相关链接:
https://blog.csdn.net/linuu/article/details/51385682
下面直接上代码,需要注意的地方,会在代码中通过注释进行说明。
使用IdleStateHandler实现心跳
下面将使用IdleStateHandler
来实现心跳,Client
端连接到Server
端后,会循环执行一个任务:随机等待几秒,然后ping
一下Server
端,即发送一个心跳包。当等待的时间超过规定时间,将会发送失败,以为Server
端在此之前已经主动断开连接了。代码如下:
Client端
ClientIdleStateTrigger —— 心跳触发器
类ClientIdleStateTrigger
也是一个Handler
,只是重写了userEventTriggered
方法,用于捕获IdleState.WRITER_IDLE
事件(未在指定时间内向服务器发送数据),然后向Server
端发送一个心跳包。
/*** <p>* 用于捕获{@link IdleState#WRITER_IDLE}事件(未在指定时间内向服务器发送数据),然后向<code>Server</code>端发送一个心跳包。* </p>*/
public class ClientIdleStateTrigger extends ChannelInboundHandlerAdapter {public static final String HEART_BEAT = "heart beat!";@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleState state = ((IdleStateEvent) evt).state();if (state == IdleState.WRITER_IDLE) {// write heartbeat to serverctx.writeAndFlush(HEART_BEAT);}} else {super.userEventTriggered(ctx, evt);}}}
Pinger —— 心跳发射器
/*** <p>客户端连接到服务器端后,会循环执行一个任务:随机等待几秒,然后ping一下Server端,即发送一个心跳包。</p>*/
public class Pinger extends ChannelInboundHandlerAdapter {private Random random = new Random();private int baseRandom = 8;private Channel channel;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {super.channelActive(ctx);this.channel = ctx.channel();ping(ctx.channel());}private void ping(Channel channel) {int second = Math.max(1, random.nextInt(baseRandom));System.out.println("next heart beat will send after " + second + "s.");ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {@Overridepublic void run() {if (channel.isActive()) {System.out.println("sending heart beat to the server...");channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);} else {System.err.println("The connection had broken, cancel the task that will send a heart beat.");channel.closeFuture();throw new RuntimeException();}}}, second, TimeUnit.SECONDS);future.addListener(new GenericFutureListener() {@Overridepublic void operationComplete(Future future) throws Exception {if (future.isSuccess()) {ping(channel);}}});}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {// 当Channel已经断开的情况下, 仍然发送数据, 会抛异常, 该方法会被调用.cause.printStackTrace();ctx.close();}
}
ClientHandlersInitializer —— 客户端处理器集合的初始化类
public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {private ReconnectHandler reconnectHandler;private EchoHandler echoHandler;public ClientHandlersInitializer(TcpClient tcpClient) {Assert.notNull(tcpClient, "TcpClient can not be null.");this.reconnectHandler = new ReconnectHandler(tcpClient);this.echoHandler = new EchoHandler();}@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));pipeline.addLast(new LengthFieldPrepender(4));pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));pipeline.addLast(new Pinger());}
}
注:上面的
Handler
集合,除了Pinger
,其他都是编解码器和解决粘包,可以忽略。
TcpClient —— TCP连接的客户端
public class TcpClient {private String host;private int port;private Bootstrap bootstrap;/** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */private Channel channel;public TcpClient(String host, int port) {this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));}public TcpClient(String host, int port, RetryPolicy retryPolicy) {this.host = host;this.port = port;init();}/*** 向远程TCP服务器请求连接*/public void connect() {synchronized (bootstrap) {ChannelFuture future = bootstrap.connect(host, port);this.channel = future.channel();}}private void init() {EventLoopGroup group = new NioEventLoopGroup();// bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new ClientHandlersInitializer(TcpClient.this));}public static void main(String[] args) {TcpClient tcpClient = new TcpClient("localhost", 2222);tcpClient.connect();}}
Server端
ServerIdleStateTrigger —— 断连触发器
/*** <p>在规定时间内未收到客户端的任何数据包, 将主动断开该连接</p>*/
public class ServerIdleStateTrigger extends ChannelInboundHandlerAdapter {@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleState state = ((IdleStateEvent) evt).state();if (state == IdleState.READER_IDLE) {// 在规定时间内没有收到客户端的上行数据, 主动断开连接ctx.disconnect();}} else {super.userEventTriggered(ctx, evt);}}
}
ServerBizHandler —— 服务器端的业务处理器
/*** <p>收到来自客户端的数据包后, 直接在控制台打印出来.</p>*/
@ChannelHandler.Sharable
public class ServerBizHandler extends SimpleChannelInboundHandler<String> {private final String REC_HEART_BEAT = "I had received the heart beat!";@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String data) throws Exception {try {System.out.println("receive data: " + data);
// ctx.writeAndFlush(REC_HEART_BEAT);} catch (Exception e) {e.printStackTrace();}}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("Established connection with the remote client.");// do somethingctx.fireChannelActive();}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println("Disconnected with the remote client.");// do somethingctx.fireChannelInactive();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}
ServerHandlerInitializer —— 服务器端处理器集合的初始化类
/*** <p>用于初始化服务器端涉及到的所有<code>Handler</code></p>*/
public class ServerHandlerInitializer extends ChannelInitializer<SocketChannel> {protected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(5, 0, 0));ch.pipeline().addLast("idleStateTrigger", new ServerIdleStateTrigger());ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));ch.pipeline().addLast("decoder", new StringDecoder());ch.pipeline().addLast("encoder", new StringEncoder());ch.pipeline().addLast("bizHandler", new ServerBizHandler());}}
注:
new IdleStateHandler(5, 0, 0)
该handler
代表如果在5秒内没有收到来自客户端的任何数据包(包括但不限于心跳包),将会主动断开与该客户端的连接。
TcpServer —— 服务器端
public class TcpServer {private int port;private ServerHandlerInitializer serverHandlerInitializer;public TcpServer(int port) {this.port = port;this.serverHandlerInitializer = new ServerHandlerInitializer();}public void start() {EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(this.serverHandlerInitializer);// 绑定端口,开始接收进来的连接ChannelFuture future = bootstrap.bind(port).sync();System.out.println("Server start listen at " + port);future.channel().closeFuture().sync();} catch (Exception e) {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();e.printStackTrace();}}public static void main(String[] args) throws Exception {int port = 2222;new TcpServer(port).start();}
}
至此,所有代码已经编写完毕。
测试
首先启动客户端,再启动服务器端。启动完成后,在客户端的控制台上,可以看到打印如下类似日志:
在服务器端可以看到控制台输出了类似如下的日志:
可以看到,客户端在发送4个心跳包后,第5个包因为等待时间较长,等到真正发送的时候,发现连接已断开了;而服务器端收到客户端的4个心跳数据包后,迟迟等不到下一个数据包,所以果断断开该连接。
异常情况
在测试过程中,有可能会出现如下情况:
出现这种情况的原因是:在连接已断开的情况下,仍然向服务器端发送心跳包。虽然在发送心跳包之前会使用channel.isActive()
判断连接是否可用,但也有可能上一刻判断结果为可用,但下一刻发送数据包之前,连接就断了。
目前尚未找到优雅处理这种情况的方案,各位看官如果有好的解决方案,还望不吝赐教。拜谢!!!
断线重连
断线重连这里就不过多介绍,相信各位都知道是怎么回事。这里只说大致思路,然后直接上代码。
实现思路
客户端在监测到与服务器端的连接断开后,或者一开始就无法连接的情况下,使用指定的重连策略进行重连操作,直到重新建立连接或重试次数耗尽。
对于如何监测连接是否断开,则是通过重写ChannelInboundHandler#channelInactive
来实现,但连接不可用,该方法会被触发,所以只需要在该方法做好重连工作即可。
代码实现
注:以下代码都是在上面心跳机制的基础上修改/添加的。
因为断线重连是客户端的工作,所以只需对客户端代码进行修改。
重试策略
RetryPolicy —— 重试策略接口
public interface RetryPolicy {/*** Called when an operation has failed for some reason. This method should return* true to make another attempt.** @param retryCount the number of times retried so far (0 the first time)* @return true/false*/boolean allowRetry(int retryCount);/*** get sleep time in ms of current retry count.** @param retryCount current retry count* @return the time to sleep*/long getSleepTimeMs(int retryCount);
}
ExponentialBackOffRetry —— 重连策略的默认实现
/*** <p>Retry policy that retries a set number of times with increasing sleep time between retries</p>*/
public class ExponentialBackOffRetry implements RetryPolicy {private static final int MAX_RETRIES_LIMIT = 29;private static final int DEFAULT_MAX_SLEEP_MS = Integer.MAX_VALUE;private final Random random = new Random();private final long baseSleepTimeMs;private final int maxRetries;private final int maxSleepMs;public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries) {this(baseSleepTimeMs, maxRetries, DEFAULT_MAX_SLEEP_MS);}public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs) {this.maxRetries = maxRetries;this.baseSleepTimeMs = baseSleepTimeMs;this.maxSleepMs = maxSleepMs;}@Overridepublic boolean allowRetry(int retryCount) {if (retryCount < maxRetries) {return true;}return false;}@Overridepublic long getSleepTimeMs(int retryCount) {if (retryCount < 0) {throw new IllegalArgumentException("retries count must greater than 0.");}if (retryCount > MAX_RETRIES_LIMIT) {System.out.println(String.format("maxRetries too large (%d). Pinning to %d", maxRetries, MAX_RETRIES_LIMIT));retryCount = MAX_RETRIES_LIMIT;}long sleepMs = baseSleepTimeMs * Math.max(1, random.nextInt(1 << retryCount));if (sleepMs > maxSleepMs) {System.out.println(String.format("Sleep extension too large (%d). Pinning to %d", sleepMs, maxSleepMs));sleepMs = maxSleepMs;}return sleepMs;}
}
ReconnectHandler—— 重连处理器
@ChannelHandler.Sharable
public class ReconnectHandler extends ChannelInboundHandlerAdapter {private int retries = 0;private RetryPolicy retryPolicy;private TcpClient tcpClient;public ReconnectHandler(TcpClient tcpClient) {this.tcpClient = tcpClient;}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("Successfully established a connection to the server.");retries = 0;ctx.fireChannelActive();}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {if (retries == 0) {System.err.println("Lost the TCP connection with the server.");ctx.close();}boolean allowRetry = getRetryPolicy().allowRetry(retries);if (allowRetry) {long sleepTimeMs = getRetryPolicy().getSleepTimeMs(retries);System.out.println(String.format("Try to reconnect to the server after %dms. Retry count: %d.", sleepTimeMs, ++retries));final EventLoop eventLoop = ctx.channel().eventLoop();eventLoop.schedule(() -> {System.out.println("Reconnecting ...");tcpClient.connect();}, sleepTimeMs, TimeUnit.MILLISECONDS);}ctx.fireChannelInactive();}private RetryPolicy getRetryPolicy() {if (this.retryPolicy == null) {this.retryPolicy = tcpClient.getRetryPolicy();}return this.retryPolicy;}
}
ClientHandlersInitializer
在之前的基础上,添加了重连处理器ReconnectHandler
。
public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {private ReconnectHandler reconnectHandler;private EchoHandler echoHandler;public ClientHandlersInitializer(TcpClient tcpClient) {Assert.notNull(tcpClient, "TcpClient can not be null.");this.reconnectHandler = new ReconnectHandler(tcpClient);this.echoHandler = new EchoHandler();}@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(this.reconnectHandler);pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));pipeline.addLast(new LengthFieldPrepender(4));pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));pipeline.addLast(new Pinger());}
}
TcpClient
在之前的基础上添加重连、重连策略的支持。
public class TcpClient {private String host;private int port;private Bootstrap bootstrap;/** 重连策略 */private RetryPolicy retryPolicy;/** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */private Channel channel;public TcpClient(String host, int port) {this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));}public TcpClient(String host, int port, RetryPolicy retryPolicy) {this.host = host;this.port = port;this.retryPolicy = retryPolicy;init();}/*** 向远程TCP服务器请求连接*/public void connect() {synchronized (bootstrap) {ChannelFuture future = bootstrap.connect(host, port);future.addListener(getConnectionListener());this.channel = future.channel();}}public RetryPolicy getRetryPolicy() {return retryPolicy;}private void init() {EventLoopGroup group = new NioEventLoopGroup();// bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new ClientHandlersInitializer(TcpClient.this));}private ChannelFutureListener getConnectionListener() {return new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (!future.isSuccess()) {future.channel().pipeline().fireChannelInactive();}}};}public static void main(String[] args) {TcpClient tcpClient = new TcpClient("localhost", 2222);tcpClient.connect();}}
测试
在测试之前,为了避开 Connection reset by peer
异常,可以稍微修改Pinger
的ping()
方法,添加if (second == 5)
的条件判断。如下:
private void ping(Channel channel) {int second = Math.max(1, random.nextInt(baseRandom));if (second == 5) {second = 6;}System.out.println("next heart beat will send after " + second + "s.");ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {@Overridepublic void run() {if (channel.isActive()) {System.out.println("sending heart beat to the server...");channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);} else {System.err.println("The connection had broken, cancel the task that will send a heart beat.");channel.closeFuture();throw new RuntimeException();}}}, second, TimeUnit.SECONDS);future.addListener(new GenericFutureListener() {@Overridepublic void operationComplete(Future future) throws Exception {if (future.isSuccess()) {ping(channel);}}});}
启动客户端
先只启动客户端,观察控制台输出,可以看到类似如下日志:
可以看到,当客户端发现无法连接到服务器端,所以一直尝试重连。随着重试次数增加,重试时间间隔越大,但又不想无限增大下去,所以需要定一个阈值,比如60s。如上图所示,当下一次重试时间超过60s时,会打印Sleep extension too large(*). Pinning to 60000
,单位为ms。出现这句话的意思是,计算出来的时间超过阈值(60s),所以把真正睡眠的时间重置为阈值(60s)。
启动服务器端
接着启动服务器端,然后继续观察客户端控制台输出。
可以看到,在第9次重试失败后,第10次重试之前,启动的服务器,所以第10次重连的结果为Successfully established a connection to the server.
,即成功连接到服务器。接下来因为还是不定时ping
服务器,所以出现断线重连、断线重连的循环。
扩展
在不同环境,可能会有不同的重连需求。有不同的重连需求的,只需自己实现RetryPolicy
接口,然后在创建TcpClient
的时候覆盖默认的重连策略即可。
完!!!
END
推荐好文
强大,10k+点赞的 SpringBoot 后台管理系统竟然出了详细教程!分享一套基于SpringBoot和Vue的企业级中后台开源项目,代码很规范!
能挣钱的,开源 SpringBoot 商城系统,功能超全,超漂亮!
服务的心跳机制与断线重连,Netty底层是怎么实现的?相关推荐
- 面试官问:服务的心跳机制与断线重连,Netty底层是怎么实现的?懵了
点击上方蓝色"方志朋",选择"设为星标" 回复"666"获取独家整理的学习资料! 心跳机制 何为心跳 所谓心跳, 即在 TCP 长连接中, ...
- Netty是如何实现TCP心跳机制与断线重连的
本文来说下Netty 是如何实现 TCP 心跳机制与断线重连的 文章目录 什么是心跳机制HeartBeat 如何实现心跳机制 Netty实现自定义的心跳机制 服务端 客户端 测试效果 客户端断线重连 ...
- WebSocket的心跳机制和断线重连
背景 在服务器重启或是弱网情况下,前端不能保证一直连接成功.因此在出现被动断开的情况下,需要有心跳机制和断线重连的功能. 心跳机制:客户端每隔一段时间向服务端发送一个特有的心跳消息,每次服务端收到消息 ...
- Netty——心跳机制与断线重连
心跳机制与断线重连 心跳机制 IdleStateHandler 客户端 服务端 测试 正常情况 异常情况 总结 断线重连 为了保证系统的稳定性,心跳机制和断线重连可是必不可少的,而这两个在Netty中 ...
- 浅析 Netty 实现心跳机制与断线重连
基础 何为心跳 顾名思义, 所谓 心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性. 为什么需要心跳 因为网络的不可 ...
- Netty 心跳机制及断线重连
1.心跳检测 心跳检测是在TCP长连接中,客户端和服务端定时向对方发送数据包通知对方自己还在线,保证连接的有效性的一种机制. 为什么使用心跳检测? 假死:如果底层的TCP连接(socket连接)已经断 ...
- Netty实现心跳机制与断线重连
点击上方蓝色"方志朋",选择"设为星标" 回复"666"获取独家整理的学习资料! 来源:https://www.jianshu.com/p/ ...
- 长连接 、短连接、心跳机制与断线重连
在不同场景下要考虑长连接还是短连接,那么我们要先了解他. 短连接 概念 client与server通过三次握手建立连接,client发送请求消息,server返回响应,一次连接就完成了. 这时候双方任 ...
- 四、Netty 实现心跳机制与断线重连
一.概述 何为心跳 顾名思义, 所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性. 为什么需要心跳 因为网络的不 ...
最新文章
- INADDR_ANY的理解
- MFC接收ShellExecute多个参数
- halcon知识:工业相机问答(更新中...)
- C++编码中减少内存缺陷的方法和工具
- Java 洛谷 P1002 过河卒讲解
- boost::geometry::closeable_view用法的测试程序
- mysql分库一致性_分库分表带来的完整性和一致性问题
- 论文浅尝 | PairRE: 通过成对的关系向量实现知识图谱嵌入
- 推荐一款eclipse快速打开项目文件夹所在路径插件
- C语言编程can通讯ab8位字符,计算机专业对口升学专业试题
- python中pandas计数_python – Pandas:根据条件计数进行分组
- [置顶] Android仿人人客户端(v5.7.1)——应用主界面之左侧面板UI实现
- 关于vs2010调用python中Py_Initialize函数报错的原因
- Internet Download Manager永久版功能强大的网络下载器
- OpenCV算法精解2--OpenCV中C++基本操作2
- 金蝶k3单据编码规则_金蝶K3ERP编码规则.doc
- MT4MT5跟单EA系统跨平台
- 成就:优秀的管理者成就自己,卓越的管理者成就他人(读后感)
- 网页调起支付宝付款和微信付款
- 如何创建 Ubuntu 信息亭(kiosk)
热门文章
- 输光200亿,利润暴跌77%!史上最“作死”老字号,还想靠天价翻身?
- 何小鹏“维权”事后谈造车:心很累 曾购上千瓶白酒缓解压力
- 交通部出手了!共享单车预付资金额度不得超过这个数
- 深入理解异步Web服务器 Tornado
- Linux下logrotate命令使用.配置和理解
- 进程的三个状态的转换
- linux中进程unit是什么意思,Linux系统之进程及服务的控制
- 删除计算机共享信息命令,win10系统使用命令将网络共享删除的操作办法
- linux保存python文件_告诉Python将.txt文件保存到Linux上的某个目录 - python
- MFC小笔记:控件随窗口变化