一、RabbitMq基础知识

0、概述

消息队列的作用就是接收消息生产者的消息，然后将消息发送到消费者

1、信道channel

我的理解是生产者/消费者和rabbitmq交互的一个通道，负责交换机、队列管理；消息发布和消费管理；事务管理等

2、交换机

四种交换机：

direct：可以用一个或者多个key绑定到一个或者多个队列上

topic：支持路由的适配符 # *

Fanout广播：将消息发送给所有的队列

Header头交换机：自定义通过头消息属性来定义路由的匹配

3、队列：保存消息的队列

4、消费者：消息的接收者

5、生产者：消息的发送者

二、 使用com.rabbitmq.client.*操作mq

2.1、基本操作

0、环境和依赖

<!-- 环境

* jdk 1.8

* idea

* springboot 2.2.6

-->

<!-- 依赖 这里只导入这个包，其中包含了Rabbit client的包-->

<dependency>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

1、创建连接和信道

//获取连接

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");//mq主机地址

factory.setPort(5672);//端口，默认时5672

factory.setUsername("leyou");

factory.setPassword("leyou");

factory.setVirtualHost("/leyou");

Connection connection = factory.newConnection();

//获取信道

Channel channel = connection..createChannel();

2、申明交换机 / 队列 / 绑定交换机和队列

//交换机名，交换机类型

channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);

/**

* 第一个参数是queue：要创建的队列名

* 第二个参数是durable：是否持久化。如果为true，可以在RabbitMQ崩溃后恢复消息

* 第三个参数是exclusive：true表示一个队列只能被一个消费者占有并消费

* 第四个参数是autoDelete：true表示服务器不在使用这个队列是会自动删除它

* 第五个参数是arguments：包括死信队列，队列的ttl

*/

channel.queueDeclare(QUEUE_ONE,true,false,false,null);

//绑定交换机和队列  队列名，交换机名，routekey

channel.queueBind(QUEUE_ONE,EXCHANGE,GIRL);

3、发布消息

//1、交换机名 2、routekey 3、mandatory强制(需要return回调时必须设置为true) 4、发布消息参数 5、消息

channel.basicPublish(EXCHANGE,GIRL,true,null,"xxx降价了".getBytes());

4、接收消息

//接收消息前也需要获取连接和channel，申明队列

//接收消息

Consumer consumer = new DefaultConsumer(channel){

@Override

public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

    	//拿到消息

		System.out.println(new String(body,"utf-8"));

	}

};

/**

* 参数说明

* 1：队列名字

* 2：是否自动应答 autoACk，为false时需要手动ack

* 3：消费者，当接收到消费者时会调用给对象中的 handleDelivery 方法

*/

channel.basicConsume(QUEUE_ONE,true,consumer);

2.2、基本应用

1、功能：

有两个人小明和小华，小明对美女感兴趣，小华对股票和没事感兴趣，使用消息队列将他们感兴趣的消息发送给他们两个

2、实现：

（1）写一个类来提供创建连接和信道；

（2）生产者（发送消息方）类发送消息

（3）消费者（接收消息）类接收消息

• 连接类

public class ConnectionUtil {

    /**

     * 使用原始的rabbitmq client api 操作mq

     */

    private static ConnectionFactory factory = new ConnectionFactory();

    private static Connection connection;

    /*

    获取连接

    注意导包：需要导client下面的包

     */

    public static Connection getConnection() throws IOException, TimeoutException {

//        factory.setHost("localhost");

//        factory.setPort(5672);

        factory.setUsername("leyou");

        factory.setPassword("leyou");

        factory.setVirtualHost("/leyou");

        connection = factory.newConnection();

        return connection;

    }

    public static void close() throws IOException {

        connection.close();

    }

    /*

    创建信道

     */

    public static Channel getChannel() throws IOException, TimeoutException {

        return getConnection().createChannel();

    }

}

• 生产者

//生产者

public class provice{

    public void producerMsg() throws IOException, TimeoutException, InterruptedException {

        Channel channel = ConnectionUtil.getChannel();

        String EXCHANGE = "direct_exchange";

        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);

        //定义两个队列名

        String QUEUE_ONE = "beauty_queue";

        String QUEUE_TWO = "food_queue";

        channel.queueDeclare(QUEUE_ONE,true,false,false,null);

        channel.queueDeclare(QUEUE_TWO,true,false,false,null);

        //定义三个key

        String GIRL = "girl";

        String SHARE = "share";

        String FOOD = "food";

        //绑定

        channel.queueBind(QUEUE_ONE,EXCHANGE,GIRL);

        channel.queueBind(QUEUE_TWO,EXCHANGE,SHARE);

        channel.queueBind(QUEUE_TWO,EXCHANGE,FOOD);

        //发送消息

        /**

         * 参数：1交换机，2routekey 3 mandatory：强制；(需要return回调时必须设置为true)

         * 3参数，4消息字节数据

         */

        channel.basicPublish(EXCHANGE,GIRL,true,null,"快看，是她".getBytes());

		channel.basicPublish(EXCHANGE,SHARE,true,null,"股票涨了".getBytes());

        channel.basicPublish(EXCHANGE,FOOD,true,null,"肯德基降价了".getBytes());

        //关闭连接

        channel.close();

        ConnectionUtil.close();

    }

}

• 消费者

public class ConsumerMq { // 消费消息

    /**

     * 使用原始的rabbitmq client api 操作mq

     */

    String EXCHANGE = "direct_exchange";

    String QUEUE_ONE = "beauty_queue";

    String QUEUE_TWO = "food_queue";

    //key

    String GIRL = "girl";

    String SHARE = "share";

    String FOOD = "food";

    public void consumer() throws IOException, TimeoutException {

        Channel channel = ConnectionUtil.getChannel();

        /**

         * 第一个参数是queue：要创建的队列名

         * 第二个参数是durable：是否持久化。如果为true，可以在RabbitMQ崩溃后恢复消息

         * 第三个参数是exclusive：true表示一个队列只能被一个消费者占有并消费

         * 第四个参数是autoDelete：true表示服务器不在使用这个队列是会自动删除它

         * 第五个参数是arguments：包括死信队列，队列的ttl，

         */

        channel.queueDeclare(QUEUE_ONE,true,false,false,null);

        channel.queueDeclare(QUEUE_TWO,true,false,false,null);

        //在生产者绑定了交换机和队列，在这里就不需要绑定

        //channel.queueBind(QUEUE_ONE,EXCHANGE,GIRL);

        //channel.queueBind(QUEUE_TWO,EXCHANGE,SHARE);

        //channel.queueBind(QUEUE_TWO,EXCHANGE,FOOD);

        //接收消息

        Consumer consumer = new DefaultConsumer(channel){

            @Override

            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.out.println(new String(body,"utf-8"));

                //手动应答ack可以在该方法中进行；参数：1.消息tag，2.是否批量ack

                channel.basicAck(envelope.getDeliveryTag(),false);

            }

        };

        /**

         * 参数说明

         * 1：队列名字

         * 2：是否自动应答 autoACk 为false时需要手动ack

         * 3：消费者，当接收到消费者时会调用给对象中的 handleDelivery 方法

         */

        channel.basicConsume(QUEUE_ONE,false,consumer);

        channel.basicConsume(QUEUE_TWO,false,consumer);

    }

}

2.3、mq事务，发送方确认，和消息回调

概述

消息的发送链路 生产者 -> exchange --> queue --> 消费者；为确保消息发送到rabbitmq，amqp协议提供了三个机制来保证：事务，发送方确认（ack），消息回调（returncallback）；事务的方式和数据库的事务类似，这里不做详细介绍；发送方确认是当消息发送到交换机时， broker（实现amqp协议的服务端，这里指rabbitmq）会回调发送者的一个固定方法来确认消息成功发送；消息回调是发生在交换机通过路由key转发到队列的过程中，如果消息不能通过key找到对应的queue则回调一个固定方法将消息返回给生产者，确保消息不丢失

1、mq事务

• rabbitMq是支持事务的，但是使用事务的效率很低，在消息数量很大的情况下影响性能

2、发送方确认

对于固定消息体大小和线程数，如果消息持久化，生产者confirm(或者采用事务机制)，消费者ack那么对性能有很大的影响.

消息持久化的优化没有太好方法，用更好的物理存储（SAS, SSD, RAID卡）总会带来改善。生产者confirm这一环节的优化则主要在于客户端程序的优化之上。归纳起来，客户端实现生产者confirm有三种编程方式：

1. 普通confirm模式：每发送一条消息后，调用waitForConfirms()方法，等待服务器端confirm。实际上是一种串行confirm了。
2. 批量confirm模式：每发送一批消息后，调用waitForConfirms()方法，等待服务器端confirm。
3. 异步confirm模式：提供一个回调方法，服务端confirm了一条或者多条消息后Client端会回调这个方法。

• [ ]  普通confirm模式

//要点

//第1种

//普通confirm模式最简单，publish一条消息后，等待服务器端confirm,如果服务端返回false或者超时时间内未返回，客户端进行消息重传。

//1.发消息前

channel.confirmSelect();

//2.发消息后

//判断消息发送是否成功

if(channel.waitForConfirms()){

    System.out.println("消息发送成功");

}

• [ ]  批量confirm模式

批量confirm模式稍微复杂一点，客户端程序需要定期（每隔多少秒）或者定量（达到多少条）或者两则结合起来publish消息，然后等待服务器端confirm, 相比普通confirm模式，批量极大提升confirm效率，但是问题在于一旦出现confirm返回false或者超时的情况时，客户端需要将这一批次的消息全部重发，这会带来明显的重复消息数量，并且，当消息经常丢失时，批量confirm性能应该是不升反降的。

channel.confirmSelect();

for(int i=0;i<batchCount;i++){

	channel.basicPublish(ConfirmConfig.exchangeName, ConfirmConfig.routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, ConfirmConfig.msg_10B.getBytes());

}

if(!channel.waitForConfirms()){

	System.out.println("send message failed.");

}

异步confirm模式

异步confirm模式的编程实现最复杂，Channel对象提供的ConfirmListener()回调方法只包含deliveryTag（当前Chanel发出的消息序号），我们需要自己为每一个Channel维护一个unconfirm的消息序号集合，每publish一条数据，集合中元素加1，每回调一次handleAck方法，unconfirm集合删掉相应的一条（multiple=false）或多条（multiple=true）记录。从程序运行效率上看，这个unconfirm集合最好采用有序集合SortedSet存储结构。实际上，SDK中的waitForConfirms()方法也是通过SortedSet维护消息序号的。

关键代码：

SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

//别忘这行代码

 channel.confirmSelect();

//添加监听器

channel.addConfirmListener(new ConfirmListener() {

    public void handleAck(long deliveryTag, boolean multiple) throws IOException {

        if (multiple) {

            confirmSet.headSet(deliveryTag + 1).clear();

        } else {

            confirmSet.remove(deliveryTag);

        }

    }

    public void handleNack(long deliveryTag, boolean multiple) throws IOException {

        System.out.println("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);

        if (multiple) {

            confirmSet.headSet(deliveryTag + 1).clear();

        } else {

            confirmSet.remove(deliveryTag);

        }

    }

});

while (true) {

    long nextSeqNo = channel.getNextPublishSeqNo();

    channel.basicPublish(ConfirmConfig.exchangeName, ConfirmConfig.routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, ConfirmConfig.msg_10B.getBytes());

    confirmSet.add(nextSeqNo);

}

3、消息回调

//要点

//1.发送消息是将第三个参数mandatory设置为true

channel.basicPublish(EXCHANGE,FOOD,true,null,"肯德基降价了".getBytes());

//2.添加消息回调监听器

channel.addReturnListener(new ReturnListener() {

     @Override

     public void handleReturn(int i, String s, String s1, String s2, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {

         System.out.println("消息不可路由"+new String(bytes,"utf-8"));

     }

 });

//注意：开启回调不能关闭连接和信道，

2.4、接收方确认

1、概述

接收方ack分为手动和自动，在接收消息时设置

//第二个参数就是指定是否手动ack false时为手动

channel.basicConsume(QUEUE_ONE,false,consumer);

手动ack有三种

• 单个确认
• 单个拒绝
• 批量拒绝

2、代码实现

单个确认ack

//接收消息

Consumer consumer = new DefaultConsumer(channel){

    @Override

    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

        System.out.println(new String(body,"utf-8"));

        //手动应答ack可以在该方法中进行；参数：1.消息tag，2.是否批量ack

		channel.basicAck(envelope.getDeliveryTag(),false);

        //拒绝消息;参数：1.消息tag；2.消息是否重新入队，当只有一个消费者时，会引起重复消费

        channel.basicReject(envelope.getDeliveryTag(),false);

        //批量ack消息;参数：1.消息tag；2.是否批量ack消息，3.是否重回队列

		channel.basicNack(envelope.getDeliveryTag(),true,false);

    }

};

//这里只需要条应答的语句，我这里知识都列出来

channel.basicConsume(QUEUE_ONE,false,consumer);

//注意上面第二个参数要为false才能手动ack

2.5、消息TTL和队列TTL、死信队列、延迟队列

这一块暂时不使用原始RabbitMq Client API实现，后面再研究，但是会使用下面的org.springframework.amqp来实现

三、使用org.springframework.amqp操作mq

3.1、前言：

Spring对RabbitMp进行了抽象，将交换机，队列，消息，绑定，连接等抽象出实体类，方便操作，还提供了RabbitAdmit 和RabbitTemplate 来方便交换机队列的管理以及消息的发送接收等

3.2、基本实例

0、环境和依赖

<!-- 环境

* jdk 1.8

* idea

* springboot 2.2.6

  -->

<!-- 依赖 -->

<dependency>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

1、实例

发送消息告诉消费者超时打折了快来购物

• 配置类

@Configuration

public class RabbitConfig {

    private final static Logger log = LoggerFactory.getLogger(RabbitConfig.class);

    private final static String EXCHANGE_NAME = "verification_code_exchange";

    private final static String VERIFICATION_CODE_QUEUE = "verification_code_queue";

    private final static String VERIFICATION_CODE_ROUTE_KEY = "verification_code_key";

    //死信交换机和队列和key

    private final static String DLX_EXCHANGE_NAME = "dlx-exchange";

    private final static String DLX_KEY = "verification_code_key";

    @Bean

    public CachingConnectionFactory connectionFactory(){

        CachingConnectionFactory conn = new CachingConnectionFactory();

        conn.setUsername("leyou");

        conn.setPassword("leyou");

        conn.setVirtualHost("/leyou");

        //消息发送到mq发送确认消息给生产者

        conn.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);

        //消息发送到mq，通过绑定的key找不到queue，则发送消息给生产者

        conn.setPublisherReturns(true);

        return conn;

    }

    @Bean

    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        //设置消息序列化

        rabbitTemplate.setMessageConverter(converter());

        //消息的确认回调

//        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

//            @Override

//            public void confirm(CorrelationData correlationData, boolean b, String s) {

//

//            }

//        });

        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {

            //ack为确认消息是否成功发送到mq

            if(ack){

                //成功发送

                log.info("消息发送成功");

            }

        });

        //改标志位设置位true时，当交换机根据自身类型和routeKey无法找到对应的队列时，

        // 则mq会将消息返还给生产者

        //当为false时则mq会将消息直接删除

        rabbitTemplate.setMandatory(true);

        //消息，返回码，返回内容，交换机，路由key

        rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText, String exchange, String routingKey)->{

            //消息

            log.info("message:{},replyCode:{},replyText:{},exchange:{},routingKey:{}",message,replyCode,replyText,exchange,routingKey);

        });

        return rabbitTemplate;

    }

    /**

     * 注入rabbitadmin 用来申明交换机和队列，主要作用是代替原始的使用channl申明的做法，全部交给这个对象来完成

     * @param connectionFactory

     * @return

     */

    @Bean

    public RabbitAdmin rabbitAdmit(CachingConnectionFactory connectionFactory){

        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);

        return rabbitAdmin;

    }

    /**

     * 消息序化对象

     * 默认使用的是JDK的序列化，这里配置了后就可以将消息序列化为json格式

     */

    @Bean

    public MessageConverter converter() {

        return new Jackson2JsonMessageConverter();

    }

    /**

     * 申明一个交换机

     */

    @Bean

    public DirectExchange verificationCodeExchange(RabbitAdmin rabbitAdmin){

        DirectExchange exchange = new DirectExchange(EXCHANGE_NAME);

        rabbitAdmin.declareExchange(exchange);

        return exchange;

    }

    /**

     * 申明一个队列

     * @param rabbitAdmin

     * @return

     */

    @Bean

    public Queue getQueue(RabbitAdmin rabbitAdmin){

        Queue queue = new Queue(VERIFICATION_CODE_QUEUE, true,false,false,null);

        rabbitAdmin.declareQueue(queue);

        return queue;

    }

    /**

     * 申明一个绑定

     * @param rabbitAdmin

     * @param verificationCodeExchange

     * @return

     */

    @Bean

    public Binding bindingQueue(RabbitAdmin rabbitAdmin,DirectExchange verificationCodeExchange){

        Binding with = BindingBuilder.bind(getQueue(rabbitAdmin)).to(verificationCodeExchange).with(VERIFICATION_CODE_ROUTE_KEY);

        rabbitAdmin.declareBinding(with);

        return with;

    }

}

说明：上面用到了生产者confirm和消息回调机制

1、生产者confirm关键代码：

//1、创建连接时

conn.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);

//2、创建rabbitTemplate时

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {

    //ack为确认消息是否成功发送到mq

    if(ack){

        //成功发送

        log.info("消息发送成功");

    }

});

2、消息回调机制关键代码：

//1、创建连接时

conn.setPublisherReturns(true);

//2、创建rabbitTemplate时

//改标志位设置位true时，当交换机根据自身类型和routeKey无法找到对应的队列时，

// 则mq会将消息返还给生产者

//当为false时则mq会将消息直接删除

rabbitTemplate.setMandatory(true);

rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText, String exchange, String routingKey)->{

            //消息

log.info("message:{},replyCode:{},replyText:{},exchange:{},routingKey:{}",message,replyCode,replyText,exchange,routingKey);

});

生产者：

@Component

public class RabbitSender {

    //注入rabbitTemplate

    @Autowired

    private RabbitTemplate rabbitTemplate;

    public void sendMsg(){

        //构建消息

        Message message = MessageBuilder.withBody(

                JSONObject.toJSONString(MessageModel.builder().id(msgId).context("超市打折，快来抢购！").build()).getBytes()).build();

        //消息持久化

        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);

        //消息的媒体类型

        message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);

        //消息的自定义关联id

        CorrelationData correlationData = new CorrelationData(String.valueOf(msgId));

        rabbitTemplate.convertAndSend(exchange,routingKey,message,new MessagePostProcessor(){

            //消息后置处理器，可以在下面这个方法中对消息进行相关属性的设置

            @Override

            public Message postProcessMessage(Message message) throws AmqpException {

                //比如可以设置上面 这些属性等

                //message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);//消息持久化问题          					                				//message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);//消息的媒体类型

                return message;

            }

        },correlationData);

    }

}

消费者

@Component

public class RabbitReceive {

    @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = EXCHANGE_NAME, type = ExchangeTypes.DIRECT),

            key = VERIFICATION_CODE_ROUTE_KEY,

            value = @Queue(value = VERIFICATION_CODE_QUEUE, autoDelete = "false"),

            ignoreDeclarationExceptions = "true"),

            concurrency = "1", // 指定监听该队列的消费者个数

            ackMode = "MANUAL"// 手动ack

    )

    public void receiveCode(Channel channel, Message msg, @Headers Map<String, Object> headers) throws IOException, InterruptedException {

        String msgId = (String) headers.get("spring_listener_return_correlation");

        long tag = msg.getMessageProperties().getDeliveryTag();

        channel.basicAck(tag, false);

    }

}

其中：发送方确认（生产者confirm）、消息回调上面代码都包含了；消费者ack则和原始方法是一样的

下面介绍消息TTL，队列TTL，死信队列，延迟队列

• 消息和队列的TTL

//消息ttl

//在构建消息时设置消息的过期时间

Message message = MessageBuilder.withBody(

                JSONObject.toJSONString(MessageModel.builder().id(msgId).context("超市打折，快来抢购！").build()).getBytes()).build();

//消息的过期时间

message.getMessageProperties().setExpiration("5000");

//队列的ttl

//在创建队列时通过参数设置

Map<String, Object> args = new HashMap<>();

//指定死信交换机

args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);

//指定死信队列的key

args.put("x-dead-letter-routing-key", DLX_KEY);

//设置队列中消息的过期时间 ms

args.put("x-message-ttl",10000);

//整个队列的过期时间，过期后整个队列会被删除

//args.put("x-expires",10000);

Queue queue = new Queue(VERIFICATION_CODE_QUEUE, true,false,false,args);

上面还包括死信队列的属性设置，和死信队列key，关于死信队列的配置，还需要配置一个死信交换机和一个死信队列；当有消息或队列的ttl过期，消息超过队列最大长度，消息被拒绝且设置不重新回队列，则消息会被转发到死信交换机，再转发到死信队列。

• 关于延迟队列的实现方法有两种

1. 使用死信队列，用一个设置了ttl的队列来存放消息，该队列不需要消费者监听，然后给该队列配置死信交换机和队列，消费者监听死信队列，这样就能达到时间达到延迟收到消息的目的
2. 使用rabbitmq插件的方式实现，这里先不写，放到下一篇笔记中

最后

感谢你看到这里，看完有什么的不懂的可以在评论区问我，觉得文章对你有帮助的话记得给我点个赞，每天都会分享java相关技术文章或行业资讯，欢迎大家关注和转发文章！

最新文章

1. 一种全新的屏幕适配方法 自动百分比适配 一切px说了算
2. 【BZOJ 1096】【ZJOI 2007】仓库建设 DP+斜率优化
3. eclipse各版本代号
4. paip兼容windows与linux的java类根目录路径的方法
5. jQuery判断某个元素是否存在某个样式
6. 匿名用户访问sharepoint2010中的列表
7. 技术不牛如何才拿到国内IT巨头的Offer
8. timesten备份和恢复
9. 使用ThinkPHP的扩展功能
10. BZOJ 3238: [Ahoi2013]差异 [后缀自动机]
11. ZooKeeper之（二）数据模型
12. css属性position: static|relative|absolute|fixed|sticky简单解析
13. Java对Excel数据处理（利用POI解析Excel）
14. vue组件化开发组件拆分原则是什么
15. es日常维护
16. node.js中process进程的概念和child_process子进程模块的使用
17. Netty通信原理
18. saltstack自动化运维系列④之saltstack的命令返回结果mysql数据库写入
19. SQLite Manager插件安装与使用（firefox）
20. centos 7部署ELK

热门文章

1. JAVA 基于Jusup爬虫
2. 在git下如何创建分支
3. 010_Markdown学习
4. Redis (总结）
5. mock.js 学习
6. nginx负载均衡常见问题配置信息
7. D. New Year Santa Network 解析(思維、DFS、組合、樹狀DP)
8. D. Kilani and the Game 解析(裸BFS、實作)
9. html2canvas.js——HTML转Canvas工具
10. Disruptor 使用简介