Redis和RabbitMQ在項目中的使用

一：Redis的使用

1.先引入pom.xml的依賴

<dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>

2.開始編碼

package me.silentdoer.redisrabbitmq;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.util.Pool;

import java.io.UnsupportedEncodingException;

/**
 * @author silentdoer
 * @version 1.0
 * @description the description
 * @date 4/11/18 9:03 PM
 */
public class Entrance {
    public static Pool<Jedis> jedisPool;

// TODO 在項目里對redis的使用其實更通過radis-cli的方式幾乎一模一樣；
    public static void main(String[] args) throws UnsupportedEncodingException {
        // 通過配置初始化Jedis連接池，類似初始化數據庫連接池的步驟
        initJedisPool();
        // 通過Jedis連接池來管理Jedis，這樣當jedis.close()時能夠復用內部的數據傳輸模塊
        Jedis jedis = jedisPool.getResource();

        // 設置db0中的key，value
        jedis.set("中國".getBytes("utf8"), "我里試試試".getBytes("utf8"));
        // 對於set字符串，最終也是通過set(SafeEncoder.encode(key), SafeEncoder.encode(value))轉換為byte數組,看了下是UTF-8
        //jedis.set("strKey", "value");
        // 這里的value最終也是通過SafeEncoder.encodeMany(strs)，將字符串數組（List類型的value）轉換為了二位字節數組
        //jedis.rpush("listKey", new String[]{"value1", "value2", "value3"});

        // 注意，set時是什么形式，get時最好就應該是什么形式
        byte[] content = jedis.get("中國".getBytes("utf8"));
        System.out.println(new String(content, "UTF8"));
        // 雖然這樣也行，但是為了未來兼容性，還是手動轉換為字節數組比較靠譜，比較后面的版本不一定就換成別的編碼了
        //String content = jedis.get("中國");
        //System.out.println(content);

        //選擇redis的內存db，這個大小是可以在配置里規定的，個人規定的是一個redis實例的db大小不能超過16
        //jedis.select(1);

        // 若此Jedis對象由JedisPool創建則close時將會設置此jedis不可用，但是回收內部的數據傳輸模塊
        jedis.close();
    }

    public static synchronized void initJedisPool(){
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxIdle(24);  // 最大空閑數（這里應該還有個設置多久后認定是空閑的）
        config.setMaxTotal(48);  // 最大連接數
        config.setMaxWaitMillis(3000);  // 當連接不夠時等待的時長
        config.setTestOnBorrow(false);
        try {
            jedisPool = new JedisPool(config, "localhost", 6379, 10000, null);  // 最后一個為auth，沒有即為null
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}

二：RabbitMQ的使用

1.引入依賴

<dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>4.1.0</version>
        </dependency>

2.生產者編碼

package me.silentdoer.redisrabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;

/**
 * @author silentdoer
 * @version 1.0
 * @description the description
 * @date 4/11/18 10:23 PM
 */
public class MqProducer {
    private static ConnectionFactory factory;

    public static void main(String[] args) throws IOException, TimeoutException {
        initFactory();
        Connection connection = factory.newConnection();
        // channel的實現方式其實就是注冊機制，即我通過某個tcp協議的socket對象，調有createChannel其實就是發送如：create channel 0的數據
        // 然后服務端收到后在相關注冊表里添加條目 0 - conn0 - user - vhost 之類的數據，然后返回true給客戶端，然后客戶端也添加本地注冊表
        // 這樣就通過每次發送數據都加一個channel從而將不同channel的數據分隔開
        Channel chann0 = connection.createChannel(1);  // TODO 此值要大於0
        Scanner scanner = new Scanner(System.in);
        String msg = "";
        while(!"exit".equals(msg)){
            msg = scanner.nextLine();
            chann0.basicPublish("", "test", null, msg.getBytes("utf8"));
        }
        // ""表示是默認的exchange(direct.default)，但是還是得顯示指定，TODO 注意不能是null
        //第三個參數props是消息包含的屬性信息。RabbitMQ的消息屬性和消息體是分開的，不像JMS消息那樣同時包含在javax.jms.Message對象中，這一點需要特別注意
        //chann0.basicPublish("", "test", null, "payload的消息".getBytes("utf8"));
        //chann0.queueDeclare("name", true, false, false, null);  // 創建一個queue（默認會創建一個default-queuename-queue的binding）
        //chann0.exchangeDeclare(..);  // 創建一個exchange，只能指定builtin的類型，即direct/fanout/header/topic。。
        //chann0.exchangeBind(..);  // 設置一個binding，將exchange和exchange綁定，類似責任鏈處理
        //chann0.queueBind(..);  // 設置一個binding，將queue和exchange綁定
        chann0.close();  // 注意，其實只是在本地和服務器的注冊表里刪除了此channel的信息，以及將此channel設置為不可用而以
        connection.close();  // 真正關閉tcp連接
    }

    public static void initFactory(){
        factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");  // default vhost is /
        factory.setConnectionTimeout(10000);
    }
}

3.消費者代碼

package me.silentdoer.redisrabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author silentdoer
 * @version 1.0
 * @description the description
 * @date 4/11/18 10:43 PM
 */
public class MqConsumer {
    private static ConnectionFactory factory;

    public static void main(String[] args) throws IOException, TimeoutException {
        initFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel(3);  // channel 不需要和producer的一樣
        // Consumer 是回調函數，即channel.basicConsume(..)獲得一條消息后會調有此觀察者的
        //handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)方法
        Consumer consumer = new DefaultConsumer(channel){
            // 默認此觀察者的命令方法是什么都不做的，要將這里獲取的值在其它地方引用只需寫個ConsumerSupport即可，然后外部consumerSupport.getBody();
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("收到一條需要我方處理的消息:" + message);
            }
        };
        while (true){
            // 注意，是channel在訂閱消息，這個訂閱和監聽器不太一樣，需要不斷的訂閱（有點像C#里的BeginAccept，獲取后需要繼續BeginAccept才行）
            channel.basicConsume("test", true, consumer);
        }
　　　　 //channel.close();
　　　　 //connection.close();
    }

    public static void initFactory(){
        factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");  // TODO 注意，生產環境消費者和生產者的賬戶是不一樣的，只不過vhost一樣，ip:port也一樣
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        factory.setConnectionTimeout(10000);
    }
}