- Java锁的逻辑(结合对象头和ObjectMonitor)
- 还在用饼状图?来瞧瞧这些炫酷的百分比可视化新图形(附代码实现)⛵
- 自动注册实体类到EntityFrameworkCore上下文,并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
gite: https://gitee.com/tickbh/wmproxy 。
github: https://github.com/tickbh/wmproxy 。
Go语言
的协程设计就非常优秀。协程跟线程类似,无需改变编程模型,同时它也跟 async
类似,可以支持大量的任务并发运行。 actor
,单元之间通过消息传递的方式进行通信和数据传递,跟分布式系统的设计理念非常相像。由于 actor
模型跟现实很贴近,因此它相对来说更容易实现,但是一旦遇到流控制、失败重试等场景时,就会变得不太好用 async
模型的问题就是内部实现机制过于复杂,对于用户来说,理解和使用起来也没有线程和协程简单。主要是等待完成状态 await
,就比如读socket数据,等待系统将数据送达再继续触发读操作的执行,从而答到无损耗的运行。 这里我们选择的是 async/await 的模式 。
poll
)时才会运行, 因此丢弃一个 future
会阻止它未来再被运行, 你可以将 Future
理解为一个在未来某个时间点被调度执行的任务。在Rust中调用异步函数没有用await会被编辑器警告,因为这不符合预期。 async
内部实现)都没有性能损耗,例如,你可以无需分配任何堆内存、也无需任何动态分发来使用 async
,这对于热点路径的性能有非常大的好处,正是得益于此,Rust 的异步编程性能才会这么高。 tokio
,官方版本的async目前的生态相对 tokio
会差许多 跟数据通讯相关的代码均放在 streams 目录下面.
center_client.rs
中的 CenterClient
表示中心客户端,提供主动连接服务端的能力并可选择为加密( TLS
)或者普通模式,并且将该客户端收发的消息转给服务端 center_server.rs
中的 CenterServer
表示中心服务端,接受中心客户端的连接,并且将信息处理或者转发 trans_stream.rs
中的 TransStream
表示转发流量端,提供与中心端绑定的读出写入功能,在代理服务器中客户端接收的连接因为无需处理任何数据,直接绑定为 TransStream
将数据完整的转发给服务端 virtual_stream.rs
中的 VirtualStream
表示虚拟端,虚拟出一个流连接,并实现AsyncRead及AsyncRead,可以和流一样正常操作,在代理服务器中服务端接收到新连接,把他虚拟成一个 VirtualStream
,就可以直接和他连接的服务器上做双向绑定。 下面展示的是http代理,通过加密TLS中的转化 。
上述过程实现了程序中实现了http的代理转发 。
以下是http内网穿透在代理中的转化 。
上述过程可以主动把公网的请求连接转发到内网,由内网提供完服务后再转发到公网的请求,从而实现内网穿透.
下面是代码类的定义 。
/// 中心客户端
/// 负责与服务端建立连接,断开后自动再重连
pub struct CenterClient {
/// tls的客户端连接信息
tls_client: Option<Arc<rustls::ClientConfig>>,
/// tls的客户端连接域名
domain: Option<String>,
/// 连接中心服务器的地址
server_addr: SocketAddr,
/// 内网映射的相关消息
mappings: Vec<MappingConfig>,
/// 存在普通连接和加密连接,此处不为None则表示普通连接
stream: Option<TcpStream>,
/// 存在普通连接和加密连接,此处不为None则表示加密连接
tls_stream: Option<TlsStream<TcpStream>>,
/// 绑定的下一个sock_map映射
next_id: u32,
/// 发送Create,并将绑定的Sender发到做绑定
sender_work: Sender<(ProtCreate, Sender<ProtFrame>)>,
/// 接收的Sender绑定,开始服务时这值move到工作协程中,所以不能二次调用服务
receiver_work: Option<Receiver<(ProtCreate, Sender<ProtFrame>)>>,
/// 发送协议数据,接收到服务端的流数据,转发给相应的Stream
sender: Sender<ProtFrame>,
/// 接收协议数据,并转发到服务端。
receiver: Option<Receiver<ProtFrame>>,
}
主要的逻辑流程,循环监听数据流的到达,同时等待多个异步的到达,这里用的是 tokio::select! 宏 。
loop {
let _ = tokio::select! {
// 严格的顺序流
biased;
// 新的流建立,这里接收Create并进行绑定
r = receiver_work.recv() => {
if let Some((create, sender)) = r {
map.insert(create.sock_map(), sender);
let _ = create.encode(&mut write_buf);
}
}
// 数据的接收,并将数据写入给远程端
r = receiver.recv() => {
if let Some(p) = r {
let _ = p.encode(&mut write_buf);
}
}
// 数据的等待读取,一旦流可读则触发,读到0则关闭主动关闭所有连接
r = reader.read(&mut vec) => {
match r {
Ok(0)=>{
is_closed=true;
break;
}
Ok(n) => {
read_buf.put_slice(&vec[..n]);
}
Err(_err) => {
is_closed = true;
break;
},
}
}
// 一旦有写数据,则尝试写入数据,写入成功后扣除相应的数据
r = writer.write(write_buf.chunk()), if write_buf.has_remaining() => {
match r {
Ok(n) => {
write_buf.advance(n);
if !write_buf.has_remaining() {
write_buf.clear();
}
}
Err(e) => {
println!("center_client errrrr = {:?}", e);
},
}
}
};
loop {
// 将读出来的数据全部解析成ProtFrame并进行相应的处理,如果是0则是自身消息,其它进行转发
match Helper::decode_frame(&mut read_buf)? {
Some(p) => {
match p {
ProtFrame::Create(p) => {
}
ProtFrame::Close(_) | ProtFrame::Data(_) => {
},
}
}
None => {
break;
}
}
}
}
下面是代码类的定义 。
/// 中心服务端
/// 接受中心客户端的连接,并且将信息处理或者转发
pub struct CenterServer {
/// 代理的详情信息,如用户密码这类
option: ProxyOption,
/// 发送协议数据,接收到服务端的流数据,转发给相应的Stream
sender: Sender<ProtFrame>,
/// 接收协议数据,并转发到服务端。
receiver: Option<Receiver<ProtFrame>>,
/// 发送Create,并将绑定的Sender发到做绑定
sender_work: Sender<(ProtCreate, Sender<ProtFrame>)>,
/// 接收的Sender绑定,开始服务时这值move到工作协程中,所以不能二次调用服务
receiver_work: Option<Receiver<(ProtCreate, Sender<ProtFrame>)>>,
/// 绑定的下一个sock_map映射,为双数
next_id: u32,
}
主要的逻辑流程,循环监听数据流的到达,同时等待多个异步的到达,这里用的是 tokio::select! 宏,select处理方法与Client相同,均处理相同逻辑,不同的是接收数据包后数据端是处理的proxy的请求,而Client处理的是内网穿透的逻辑 。
loop {
// 将读出来的数据全部解析成ProtFrame并进行相应的处理,如果是0则是自身消息,其它进行转发
match Helper::decode_frame(&mut read_buf)? {
Some(p) => {
match p {
ProtFrame::Create(p) => {
tokio::spawn(async move {
let _ = Proxy::deal_proxy(stream, flag, username, password, udp_bind).await;
});
}
ProtFrame::Close(_) | ProtFrame::Data(_) => {
},
}
}
None => {
break;
}
}
}
下面是代码类的定义 。
/// 转发流量端
/// 提供与中心端绑定的读出写入功能
pub struct TransStream<T>
where
T: AsyncRead + AsyncWrite + Unpin,
{
// 流有相应的AsyncRead + AsyncWrite + Unpin均可
stream: T,
// sock绑定的句柄
id: u32,
// 读取的数据缓存,将转发成ProtFrame
read: BinaryMut,
// 写的数据缓存,直接写入到stream下,从ProtFrame转化而来
write: BinaryMut,
// 收到数据通过sender发送给中心端
in_sender: Sender<ProtFrame>,
// 收到中心端的写入请求,转成write
out_receiver: Receiver<ProtFrame>,
}
主要的逻辑流程,循环监听数据流的到达,同时等待多个异步的到达,这里用的是 tokio::select! 宏,监听的对象有stream可读,可写,sender的写发送及receiver的可接收 。
loop {
// 有剩余数据,优先转化成Prot,因为数据可能从外部直接带入
if self.read.has_remaining() {
link.push_back(ProtFrame::new_data(self.id, self.read.copy_to_binary()));
self.read.clear();
}
tokio::select! {
n = reader.read(&mut buf) => {
let n = n?;
if n == 0 {
return Ok(())
} else {
self.read.put_slice(&buf[..n]);
}
},
r = writer.write(self.write.chunk()), if self.write.has_remaining() => {
match r {
Ok(n) => {
self.write.advance(n);
if !self.write.has_remaining() {
self.write.clear();
}
}
Err(_) => todo!(),
}
}
r = self.out_receiver.recv() => {
if let Some(v) = r {
if v.is_close() || v.is_create() {
return Ok(())
} else if v.is_data() {
match v {
ProtFrame::Data(d) => {
self.write.put_slice(&d.data().chunk());
}
_ => unreachable!(),
}
}
} else {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid frame"))
}
}
p = self.in_sender.reserve(), if link.len() > 0 => {
match p {
Err(_)=>{
return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid frame"))
}
Ok(p) => {
p.send(link.pop_front().unwrap())
},
}
}
}
下面是代码类的定义,我们并未有真实的socket,通过虚拟出的端方便后续的操作 。
/// 虚拟端
/// 虚拟出一个流连接,并实现AsyncRead及AsyncRead,可以和流一样正常操作
pub struct VirtualStream
{
// sock绑定的句柄
id: u32,
// 收到数据通过sender发送给中心端
sender: PollSender<ProtFrame>,
// 收到中心端的写入请求,转成write
receiver: Receiver<ProtFrame>,
// 读取的数据缓存,将转发成ProtFrame
read: BinaryMut,
// 写的数据缓存,直接写入到stream下,从ProtFrame转化而来
write: BinaryMut,
}
虚拟的流主要通过实现AsyncRead及AsyncWrite 。
impl AsyncRead for VirtualStream
{
// 有读取出数据,则返回数据,返回数据0的Ready状态则表示已关闭
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut [std](https://note.youdao.com/)[link](https://note.youdao.com/)::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
loop {
match self.receiver.poll_recv(cx) {
Poll::Ready(value) => {
if let Some(v) = value {
if v.is_close() || v.is_create() {
return Poll::Ready(Ok(()))
} else if v.is_data() {
match v {
ProtFrame::Data(d) => {
self.read.put_slice(&d.data().chunk());
}
_ => unreachable!(),
}
}
} else {
return Poll::Ready(Ok(()))
}
},
Poll::Pending => {
if !self.read.has_remaining() {
return Poll::Pending;
}
},
}
if self.read.has_remaining() {
let copy = std::cmp::min(self.read.remaining(), buf.remaining());
buf.put_slice(&self.read.chunk()[..copy]);
self.read.advance(copy);
return Poll::Ready(Ok(()));
}
}
}
}
impl AsyncWrite for VirtualStream
{
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<Result<usize, std::io::Error>> {
self.write.put_slice(buf);
if let Err(_) = ready!(self.sender.poll_reserve(cx)) {
return Poll::Pending;
}
let binary = Binary::from(self.write.chunk().to_vec());
let id = self.id;
if let Ok(_) = self.sender.send_item(ProtFrame::Data(ProtData::new(id, binary))) {
self.write.clear();
}
Poll::Ready(Ok(buf.len()))
}
}
至此基本几个大类已设置完毕,接下来仅需简单的拓展就能实现内网穿透功能.
最后此篇关于6.用Rust手把手编写一个wmproxy(代理,内网穿透等),通讯协议源码解读篇的文章就讲到这里了,如果你想了解更多关于6.用Rust手把手编写一个wmproxy(代理,内网穿透等),通讯协议源码解读篇的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
我只想从客户端向服务器发送数组 adc_array=[w, x, y, z]。下面是客户端代码,而我的服务器是在只接受 json 的 python 中。编译代码时我没有收到任何错误,但收到 2 条警告
我是 lua 和 Node js 的新手,我正在尝试将我正在开发的移动应用程序连接到服务器。问题是它连接到服务器,但我尝试传递的数据丢失或无法到达服务器。对我正在做的事情有什么问题有什么想法吗? th
我在这个页面上工作 http://www.haskell.org/haskellwiki/99_questions/Solutions/4 我理解每个函数的含义,看到一个函数可以像这样以多种方式定义,
我目前正在尝试将数据写入 excel 以生成报告。我可以将数据写入 csv 文件,但它不会按照我想要的顺序出现在 excel 中。我需要数据在每列的最佳和最差适应性下打印,而不是全部打印在平均值下。这
所以,我正在做一个项目,现在我有一个问题,所以我想得到你的帮助:) 首先,我已经知道如何编写和读取 .txt 文件,但我想要的不仅仅是 x.hasNext()。 我想知道如何像 .ini 那样编写、读
我正在尝试编写一个函数,该函数将返回作为输入给出的任何数字的阶乘。现在,我的代码绝对是一团糟。请帮忙。 function factorialize(num) { for (var i=num, i
这个问题已经有答案了: Check variable equality against a list of values (16 个回答) 已关闭 4 年前。 有没有一种简洁或更好的方法来编写这个条件
我对 VR 完全陌生,正在 AFrame 中为一个类(class)项目开发 VR 太空射击游戏,并且想知道 AFrame 中是否有 TDD 的任何文档/标准。有人能指出我正确的方向吗? 最佳答案 几乎
我正在尝试创建一个 for 循环,它将重现以下功能代码块,但以一种更具吸引力的方式。这是与 Soundcould 小部件 API 实现一起使用的 here on stackoverflow $(doc
我有一个非常令人困惑的问题。我正在尝试更改属性文件中的属性,但它只是没有更改... 这是代码: package config; import java.io.FileNotFoundException
我对 VR 完全陌生,正在 AFrame 中为一个类(class)项目开发 VR 太空射击游戏,并且想知道 AFrame 中是否有 TDD 的任何文档/标准。有人能指出我正确的方向吗? 最佳答案 几乎
我正在开发一个用户模式(Ring3)代码级调试器。它还应支持.NET可执行文件的本机(x86)调试。基本上,我需要执行以下操作: 1).NET在隐身模式下加载某些模块,而没有LOAD_DLL_DEBU
我有一个列表,我知道有些项目是不必要打印的,我正在尝试通过 if 语句来做到这一点...但是它变得非常复杂,所以有没有什么方法可以在 if 语句中包含多个索引而无需打印重写整个声明。 看起来像这样的东
我很好奇以不同方式编写 if 语句是否会影响程序的速度和效率。所以,例如写一个这样的: bool isActive = true; bool isResponding = false; if (isA
我在搜索网站的源代码时找到了一种以另一种方式(我认为)编写 if 语句的方法。 代替: if(a)b; 或: a?b:''; 我读了: !a||b; 第三种方式和前两种方式一样吗?如果是,为什么我们要
我的数据采用以下格式(HashMap的列表) {TeamName=India, Name=Sachin, Score=170} {TeamName=India, Name=Sehwag, Score=
我目前正在完成 More JOIN operations sqlzoo 的教程,遇到了下面的代码作为#12 的答案: SELECT yr,COUNT(title) FROM movie JOIN ca
我正试图找到一种更好的方法来编写这段代码: def down_up(array, player) 7.downto(3).each do |row| 8.times do |col
出于某种原因,我的缓冲区中充满了乱码,我不确定为什么。我什至用十六进制编辑器检查了我的文件,以验证我的字符是否以 2 字节的 unicode 格式保存。我不确定出了什么问题。 [打开文件] fseek
阅读编码恐怖片时,我刚刚又遇到了 FizzBuzz。 原帖在这里:Coding Horror: Why Can't Programmers.. Program? 对于那些不知道的人:FizzBu
我是一名优秀的程序员,十分优秀!