我们在很多场景下会碰到java包冲突的问题：
最出名的解决路数还是类加载机制，诸如flink，osgi都给我们提供了很多方案，这些方案都非常重型。在代码可信任的情况下，其中有一个很轻量级的解决方案就是maven-shade包。
举个例子，比方说我想在java agent中打印日志，但是又不希望和业务代码中的log4j等冲突，agent里依赖的pom文件是这样子的:
1 | <dependencies> |
这里我们log4j,slf4j可能用的版本太高或者太低，我们就可以通过打shade包的方式修改log4j和slf4j的包名,避免和业务冲突
1 | <plugin> |
通过上面的配置，artifactSet选择要修改的pom依赖，通过relocation修改包名，达到不冲突的效果。mvn clean package 后查看效果
可以发现，包名已经被修改完成,达到了避免冲突的目的。
ThreadLocal一般在多线程环境用来保存当前线程的数据。用户可以很方便地使用，并且不关心、不感知多线程的问题。下面我会用两个场景来展示多线程的问题：
测试代码分别用于ThreadLocal和FastThreadLocal。 代码如下:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import java.util.concurrent.CountDownLatch;
/**
* @author hezhangjian
*/
public class ThreadLocalTest {
public void testThreadLocal() throws Exception {
CountDownLatch cdl = new CountDownLatch(10000);
ThreadLocal<String> threadLocal = new ThreadLocal<String>();
long starTime = System.currentTimeMillis();
for (int i = 0; i < 10000; i++) {
new Thread(new Runnable() {
public void run() {
threadLocal.set(Thread.currentThread().getName());
for (int k = 0; k < 100000; k++) {
threadLocal.get();
}
cdl.countDown();
}
}, "Thread" + (i + 1)).start();
}
cdl.await();
System.out.println(System.currentTimeMillis() - starTime + "ms");
}
}
上述的代码创建了一万个线程，并将线程名设置在ThreadLocal中，随后获取这个值十万次，然后通过CountDownLoatch计算总耗时。运行这个程序大概耗时1000ms。
接下来，测试FastThreadLocal，代码基本上相似:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import io.netty.util.concurrent.FastThreadLocal;
import io.netty.util.concurrent.FastThreadLocalThread;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import java.util.concurrent.CountDownLatch;
/**
* @author hezhangjian
*/
public class FastThreadLocalTest {
public void testFastThreadLocal() throws Exception {
CountDownLatch cdl = new CountDownLatch(10000);
FastThreadLocal<String> threadLocal = new FastThreadLocal<String>();
long starTime = System.currentTimeMillis();
for (int i = 0; i < 10000; i++) {
new FastThreadLocalThread(new Runnable() {
public void run() {
threadLocal.set(Thread.currentThread().getName());
for (int k = 0; k < 100000; k++) {
threadLocal.get();
}
cdl.countDown();
}
}, "Thread" + (i + 1)).start();
}
cdl.await();
System.out.println(System.currentTimeMillis() - starTime);
}
}
跑完之后，用时还是差不多1000ms。这证明了两者在这个场景下没有什么差别
先看ThreadLocal的:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import java.util.concurrent.CountDownLatch;
/**
* @author hezhangjian
*/
public class ThreadLocalSingleThreadTest {
public void testThreadLocal() throws Exception {
CountDownLatch cdl = new CountDownLatch(1);
int size = 10000;
ThreadLocal<String> tls[] = new ThreadLocal[size];
for (int i = 0; i < size; i++) {
tls[i] = new ThreadLocal<String>();
}
new Thread(new Runnable() {
public void run() {
long starTime = System.currentTimeMillis();
for (int i = 0; i < size; i++) {
tls[i].set("value" + i);
}
for (int i = 0; i < size; i++) {
for (int k = 0; k < 100000; k++) {
tls[i].get();
}
}
System.out.println(System.currentTimeMillis() - starTime + "ms");
cdl.countDown();
}
}).start();
cdl.await();
}
}
上述的代码创建了一万个ThreadLocal，然后设置一个值，随后获取十万次数值，大概耗时2000ms
接下来我们测试FastThreadLocal
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
int size = 10000;
FastThreadLocal<String> tls[] = new FastThreadLocal[size];
for (int i = 0; i < size; i++) {
tls[i] = new FastThreadLocal<String>();
}
new FastThreadLocalThread(new Runnable() {
@Override
public void run() {
long starTime = System.currentTimeMillis();
for (int i = 0; i < size; i++) {
tls[i].set("value" + i);
}
for (int i = 0; i < size; i++) {
for (int k = 0; k < 100000; k++) {
tls[i].get();
}
}
System.out.println(System.currentTimeMillis() - starTime + "ms");
}
}).start();
}
运行结果大概只有30ms; 可以发现存在了数量级的差距。接下来重点分析ThreadLocal的机制和FastThreadLocal为什么比ThreadLocal快
我们经常会使用到set和get方法，我们分别查看一下源代码:
2
3
4
5
6
7
8
9
10
11
12
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
}
ThreadLocalMap getMap(Thread t) {
return t.threadLocals;
}
首先，获取当前的线程，然后获取存储在当前线程中的ThreadLocal变量。变量其实是一个ThreadLocalMap。最后，查看ThreadLocalMap是否为空，如果为空，则创建一个新的空Map，如果key不为空，则以ThreadLocal为key，存储这个数据
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
// We don't use a fast path as with get() because it is at
// least as common to use set() to create new entries as
// it is to replace existing ones, in which case, a fast
// path would fail more often than not.
Entry[] tab = table;
int len = tab.length;
int i = key.threadLocalHashCode & (len-1);
for (Entry e = tab[i];
e != null;
e = tab[i = nextIndex(i, len)]) {
ThreadLocal<?> k = e.get();
if (k == key) {
e.value = value;
return;
}
if (k == null) {
replaceStaleEntry(key, value, i);
return;
}
}
tab[i] = new Entry(key, value);
int sz = ++size;
if (!cleanSomeSlots(i, sz) && sz >= threshold)
rehash();
}
一般来说，ThreadLocal Map使用数组来存储数据，类似于HashMap。 每个ThreadLocal在初始化时都会分配一个threadLocal HashCode，然后按照数组的长度执行模块化操作，因此会发生哈希冲突。 在HashMap中，使用数组+链表来处理冲突，而在ThreadLocal Map中，也是一样的。 Next索引用于执行遍历操作，这显然具有较差的性能。 让我们再次看一下get方法。
2
3
4
5
6
7
8
9
10
11
12
13
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null) {
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
T result = (T)e.value;
return result;
}
}
return setInitialValue();
}
同样，首先获取当前线程，然后获取当前线程中的ThreadLocal映射，然后以当前ThreadLocal作为键来获取ThreadLocal映射中的值：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
int i = key.threadLocalHashCode & (table.length - 1);
Entry e = table[i];
if (e != null && e.get() == key)
return e;
else
return getEntryAfterMiss(key, i, e);
}
private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
Entry[] tab = table;
int len = tab.length;
while (e != null) {
ThreadLocal<?> k = e.get();
if (k == key)
return e;
if (k == null)
expungeStaleEntry(i);
else
i = nextIndex(i, len);
e = tab[i];
}
return null;
}
在相同的设置模式下，数组下标通过模块化获取来获取，否则，如果没有冲突，将遍历数据，因此可以通过分析大致了解以下问题：
译者说：为什么set的时候不适用fastPath()，因为往往大家使用完ThreadLocal都会remove，这个时候，经常是createEntry，而非updateEntry
Netty分别提供了两类FastThreadLocal和FastThreadLocalThread。 FastThreadLocalThread继承自Thread。 以下也是常用的set和get方法的源代码分析：
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
if (value != InternalThreadLocalMap.UNSET) {
set(InternalThreadLocalMap.get(), value);
} else {
remove();
}
}
public final void set(InternalThreadLocalMap threadLocalMap, V value) {
if (value != InternalThreadLocalMap.UNSET) {
if (threadLocalMap.setIndexedVariable(index, value)) {
addToVariablesToRemove(threadLocalMap, this);
}
} else {
remove(threadLocalMap);
}
}
首先，将值确定为Internal ThreadLocalMap。 UNSET，然后内部ThreadLocalMap也用于存储数据:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Thread thread = Thread.currentThread();
if (thread instanceof FastThreadLocalThread) {
return fastGet((FastThreadLocalThread) thread);
} else {
return slowGet();
}
}
private static InternalThreadLocalMap fastGet(FastThreadLocalThread thread) {
InternalThreadLocalMap threadLocalMap = thread.threadLocalMap();
if (threadLocalMap == null) {
thread.setThreadLocalMap(threadLocalMap = new InternalThreadLocalMap());
}
return threadLocalMap;
}
可以发现内部ThreadLocal映射也存储在FastThreadLocalThread中。 不同之处在于，它直接使用FastThreadLocal的index属性，而不是使用ThreadLocal的相应哈希值对位置进行建模。 实例化时初始化索引：
2
3
4
5
public FastThreadLocal() {
index = InternalThreadLocalMap.nextVariableIndex();
}
Then enter the nextVariableIndex method:
2
3
4
5
6
7
8
9
10
public static int nextVariableIndex() {
int index = nextIndex.getAndIncrement();
if (index < 0) {
nextIndex.decrementAndGet();
throw new IllegalStateException("too many thread-local indexed variables");
}
return index;
}
内部ThreadLocal映射中有一个静态nextIndex对象，用于生成数组下标，因为它是静态的，所以每个FastThreadLocal生成的索引都是连续的。 让我们看看如何在内部ThreadLocal映射中设置索引变量：
2
3
4
5
6
7
8
9
10
11
Object[] lookup = indexedVariables;
if (index < lookup.length) {
Object oldValue = lookup[index];
lookup[index] = value;
return oldValue == UNSET;
} else {
expandIndexedVariableTableAndSet(index, value);
return true;
}
}
索引变量是存储值s的对象数组； 直接使用index作为数组下标进行存储； 如果index大于数组的长度，则将其展开； get方法通过FastThreadLocal中的索引快速读取：
2
3
4
5
6
7
8
9
10
11
12
13
Object v = threadLocalMap.indexedVariable(index);
if (v != InternalThreadLocalMap.UNSET) {
return (V) v;
}
return initialize(threadLocalMap);
}
public Object indexedVariable(int index) {
Object[] lookup = indexedVariables;
return index < lookup.length? lookup[index] : UNSET;
}
通过下标直接阅读非常快，这是牺牲空间换来的速度
通过以上分析，我们可以知道，当有很多ThreadLocal读写操作时，我们可能会遇到性能问题； 另外，FastThreadLocal实现了O（1）通过空间读取数据的时间； 还有一个问题，为什么不直接使用HashMap（数组+黑红树林）代替ThreadLocalMap。
今天给大家介绍三种常见的四层负载均衡、网络转发方案，可用于四层的网关建设。
该方案需要后端服务器与前端client网络打通，GatewayIp可以采用主备的方式保证高可用
配置都在GatewayIp上，需要配置的如下:
1 | ipvsadm -A -u $GatewayIp:$port -s rr -p 600 |
如果您不希望后端Server与客户端面对面打通，那么您可能会喜欢这种方式，将GatewayIP设置为ServerIp的默认网关，再由Snat转换将报文转换出去，这样子Server就不需要与客户端面对面打通了，图示如下:
配置默认路由也很简单
1 | ip route add 客户端IP网段 via GateWayIp dev eth0 |
配置iptables
1 | iptables -t nat -A POSTROUTING -m iprange -p udp --dst-range $client_ip_range -o eth1 -j SNAT --to-source $GateWayIp |
默认路由有一个限制，就是说Server与Gateway都在一个子网内，有过商用经验的大家都知道DMZ之类的说法，就是说应用服务器和网关服务器在诸如安全组，子网等等上需要隔离。假设你需要将应用服务器和网关放在不同的子网，上面的方案就搞不定啊，这个时候需要使用ip隧道的方式来跨子网，图示如下，仅仅后边红色路线的ip发生了变化，原来的报文被ip隧道Wrap:
配置ip 隧道倒也不难
1 | ip tunnel add $tun_name mode ipip remote $remote_ip local $local_ip ttl 255 |
以上三种方案均没有单点问题，且都兼容tcp，udp协议。GateWay处的单点问题，通过zk选主、etcd选主，keepalive等 + 浮动IP迁移的方式均能解决。大家可以根据自己的网规网设自由选择
SNI是一个TLS的扩展字段，经常用于访问域名跳转到不同的后端地址。
配置方式如下：打开nginx.conf文件，以ttbb/nginx:nake镜像为例/usr/local/openresty/nginx/conf/nginx.conf
如下为默认的nginx.conf配置
1 |
|
在最后面添加上
1 | stream { |
这个时候，我们已经开启了SNI转发的功能，如果你使用backend.example.com的域名访问服务器，就会转发到backend，如果使用其他域名，就会转发到backend2
测试的时候，让我们在/etc/hosts里进行设置，添加
1 | 127.0.0.1 backend.example.com |
然后进行请求
1 | curl https://backend.example.com:12346 |
这里注意请求要使用https，http协议或者是tcp可没有SNI的说法
发现请求的确实是backend
然后测试请求127.0.0.1:12346
1 | curl https://127.0.0.1:12346 |
1 | docker run -p 2181:2181 -d ttbb/zookeeper:stand-alone |
1 | https://github.com/hezhangjian/maven-demo/tree/master/demo-zookeeper/src/main/java/com/github/hezhangjian/demo/zookeeper |
创建一个临时有序Znode，程序维持一个小时
1 | package com.github.hezhangjian.demo.zookeeper; |
创建一个临时有序Znode，创建完成后立刻关闭
1 | package com.github.hezhangjian.demo.zookeeper; |
因为是临时节点，所以存在一会后删除
也就是意味着，zk的临时node序号添加是根据父目录下一个标志计数的
该特性在Linux4.4版本引入，一个难以被大家发现的好处是，基于源地址路由，没有做地址转换，并不会在nf_conntrack中添加记录
它是对源IP和目标IP进行哈希处理(端口不参与哈希的计算)计算选路。配置的命令如下:
weight代表权重
1 | ip route add default proto static scope global \ |
1 | ip route add default proto static scope global \ |
在分布式微服务场景下，有太多的环节可以引发错误的处理（包括丢失或者重复处理），如果业务本身有幂等的特性，我们可以以较低的代价解决大部分问题。
我们假设我们在k8s集群中维护着下面的系统，部署了数个网关实例，数个业务处理服务，一套Kafka集群，数个消费者服务，一个数据库，平时的业务流程是这样子的：
1 | 客户--->(step1) 网关--->(step2) 业务处理服务-->(step3)消息中间件如Kafka-->(step4)消费者消费-->(step5) 写入数据库 |
而我们希望做到的是，在海量数据量的情况下，仅仅付出较小的代价，使得客户的每一条消息都存入到我们的数据库，没有丢失或者重复。
在分布式的环境下，有很多地方都会出错，会引发消息的丢失或者重复，我们先假设上述系统没有做太多的可靠性加固，是如下工作的：
1 | 1. 客户简单地发送到网关 |
举几个例子，那么我们会碰到类似这样的问题:
1 | 1. 客户发送到网关不考虑结果，即时我们返回失败，也不处理。显然会丢失消息，我们的系统并不能保证100%的消息处理成功 |
我们先不考虑客户，先看平台侧，我们构筑了一个什么样的系统呢？
它既不能保证至少发一次(AT_LEAST_ONCE),也不能保证最多发一次(AT_MOST_ONCE)，我们构筑了一个即可能多发也有可能少发的系统，当然，现在的基础设施很好，照着这样跑，可能丢失的数据也就是万中一二，但我们做技术，还是要有点追求，实现别人难以实现的事情才是我们的竞争力。
实现端到端一次的技术核心是两阶段提交，以上述业务流程为例，实现了两阶段提交的流程应该是这样的:
1 | 1. 消费者从Kafka消费到一批数据，但并不commit提交 |
按照这个流程，还会有一些异常场景，比如
1 | 1.消费者向Kafka提交Offset后，突然宕机，重启的消费者无法恢复事务，消息丢失 |
第二种情况属于极端场景，因为我们执行的业务简单，只是insert操作，prepare成功，commit失败，可以打日志，报告警人工处理。但第一种情况是需要自动化解决的，因为我们不能对每条提交失败的事务都人工处理。那么我们需要的是：
我们需要消费者在进程级失败的时候，可以判断处于prepare阶段的事务是否需要恢复，这个可以对比Mysql，Mysql也使用了两阶段提交协议，每次重启的时候会判断redolog处于prepare状态，binlog是否完整，如果binlog完整，则恢复。这里redo log就像我们的数据库，binlog就是kafka。但我们现在的业务没有主键，每条消息都是独立的，我们无法区分，那些是要被正常放弃的事务，那些是重启的时候需要恢复的事务。这种情况我们无法处理。
但如果这时候我们有了主键，我们有了幂等，我们只要做到至少一次就可以保证平台达到端到端一次。因为多次向数据库插入相同的数据，并不会发生什么事。
实现至少一次的核心技术是如果处理不成功就打死不提交，报告警等人工操作都不提交! 其他队列，缓存区，滑窗只不过是提升性能的手段。
1 | 1. 消费者从Kafka消费到一批数据，但并不commit提交 |
接下来我们来看与客户的通信，这次我们先看如何做到至少一次
前面说到客户可能不处理你的返回值，碰到这样的客户其实是你赚了，客户的系统连这个都不重视，那想必也不会在意你是否做到了端到端一次吧，丢失了数据想必也不是特别在意。一个端到端一次的系统，一定需要输入端和输出端的配合，正如我前面举例:
1 | 客户发送到网关不考虑结果，即时我们返回失败，也不处理。显然会丢失消息， |
那么我们也需要客户的配合，客户检测到我们回复的结果是失败，重试一下，确保成功。现在我们已经实现了至少一次了吗？
没有，还记得我前面说的这个吗？
1 | 3. 后端处理服务接收到请求，做简单校验后直接返回成功/失败，异步发送消息到Kafka |
我们需要后端处理服务接收到请求后，确保发送Kafka成功，再回复用户成功
假设前面的方案我都已经做了，在什么情况下我们会重复呢？
1 | 1. 网关在与业务处理服务的TCP的四次挥手阶段处理异常，导致业务处理成功(已发送到Kafka)，但实际返回客户失败。导致消息重复 |
为了实现两阶段提交协议，客户在发送前需要确认这个消息是否已经处理过了，但是没有主键，我们无法提供给客户这样子的信息。（如果考虑性能的话，整个系统端到端的事务，基本与高吞吐无缘了）
而且，两阶段提交协议也需要客户做很多的工作，实际中也很难落地。
常见的微服务开发下,可能会对微服务的日志打印有一些要求.常见的一种模式是,将日志文件挂载在主机路径中,然后在主机上启动filebeat收集日志.
以我的一个demo工程rabbitmq-adapt为例:
1 | 容器内打印路径: /opt/sh/logs/file.log |
但是这样子,如果我们需要在一个虚拟机上跑两个rabbitmq-adapt的时候,日志路径会重复,这种情况下,不仅我们在虚拟机很难定位问题,而且filebeat也很难给日志标记上不同实例的标签.
所以首先,我们得把日志路径按pod实例的方式隔离,选择k8s中HOSTNAME环境变量作为文件路径前缀是一个不错的选择
1 | akka@AkkadeMacBook-Pro ~ % kubectl exec -it rabbitmq-58c5f4ff6b-zthg4 bash |
这个时候,你的映射关系是 /opt/sh/logs <==> /opt/log/rabbitmq/${POD_NAME}
1 | apiVersion: apps/v1 |
可以达到如下的效果:
1 | minikube/ rabbitmq-6df8f7565c-kh2h6/ rabbitmq-6df8f7565c-ss92l/ |
这个时候,你的文件映射关系还是 /opt/sh/logs <==> /opt/log/rabbitmq
但是真正在打印日志的时候,把日志都打印到/opt/sh/logs/${POD_NAME}下,这样子也可以在一台vm上跑多个实例
1 |
|
我们解决了两个容器在一个vm上打印日志的问题,紧接着,我们要分析filebeat的能力,filebeat能否区分这两个路径,把这两个路径打上不同的标签
运行并配置好filebeat,配置文件如下
1 | filebeat.inputs: |
,查询收集上来的数据
1 | curl 127.0.0.1:9200/filebeat-7.5.1-2020.01.23-000001/_search?pretty |
1 | curl -X PUT "localhost:9200/_ingest/pipeline/attach_instance?pretty" -H 'Content-Type: application/json' -d' |
然后修改filebeat配置文件
1 | filebeat.inputs: |
查询结果就会出现在instanceId字段
三种方案均可以方便地实现同一vm上部署两个容器. 方案一只需要修改tosca模板,但要1.14版本才支持.方案二需要修改少量代码.
但前两种均不适合配置了hostnetwork,即独占主机的网络,原因,主机网络独占之后,两个容器的hostname都相同,实际中,已经独占网络的容器,还需要部署在一个节点的需求,应该比较少.
如果有,可以使用在es处处理文件路径,加上instanceId字段
在业务场景中，当需要利用ping命令对主机进行心跳探测时，直接在代码中fork进程执行ping命令虽然可行，但这种方法开销较大，并且处理流程易出错，与使用标准库相比缺乏优雅性。因此，本文探讨了使用Java的InetAddress类的isReachable方法作为替代方案。
根据资料指出，Java的InetAddress类在root用户权限下通过执行ping命令进行探测，在非root用户权限下则通过访问TCP端口7进行探测。为验证这一点，本文撰写了相应的demo代码并进行了测试（详见：GitHub - heart-beat）。
1 | import lombok.extern.slf4j.Slf4j; |
java程序打印结果也是true
此时可以看到,我们的客户端程序向目标tcp7端口发送了一个报文,虽然java程序打印结果为true,但是因为收到了RST包导致的.在当今的网络安全要求下,7端口往往不会开放
1 | iptables -A INPUT -p tcp --dport 7 -j DROP |
发送的报文没有收到RST包,此时java程序返回false.不是我们预期的结果
进一步的研究发现，Java程序发送ping命令需要创建raw socket，这要求程序具有root权限或cap_net_raw权限。赋予Java程序创建raw socket的权限后重新测试，发现程序能够正确发送ping命令，达到预期效果。
1 | setcap cap_net_raw+ep /usr/java/jdk-13.0.1/bin/java |
发现如下报错
1 | java: error while loading shared libraries: libjli.so: cannot open shared object file: No such file or directory |
随后抓包,发现还是发送了ping命令,达到了我们预期的效果
本文通过一系列测试得出结论，root用户权限下的Java程序会使用ping命令进行探测。若普通用户不具备相应权限，则会尝试探测TCP端口7，但在安全组未开启该端口的情况下会导致预期结果不一致。推荐赋予java程序特权,使得InetAddress类能够使用ping命令进行探测
或者按住 CTRL + F，输入显示过滤器
frame.len<=128
!arp
ip.addr==192.168.0.1
!tcp.port==3389
tcp.flags.syn==1
tcp.flags.rst==1
tcp.analysis.ack_rtt > 0.2 and tcp.len == 0
###启用TCP Relative Sequence Number的情况
如何启用?
Edit -> Preferences -> Protocols -> TCP Relative Sequence Numbers
tcp.flags.reset == 1 && tcp.seq == 1
tcp.flags.syn == 1 && tcp.analysis.retransmission
tcp contains {str}
http
tcp.port == 23 || tcp.port == 21
email || pop || imap
http.host == <”hostname”>
http.request.method == ‘GET’
http.request.uri == <”Full request URI”>
http matches “.zip” && http.request.method == ‘GET’