【并发】6、借助FQueue 实现多线程生产消费队列

1.这里先要说一下为什么会想到fqueue，因为这个是一个轻量级的消息队列框架，并且速度很快，用起来很方便，就是这样

当然后期考虑使用redis，这里先上一个fqueue的版本，后面有时间我再吧他改成redis版本吧，感觉可能redis版本可能更适合

package queue.fqueue.vo;

/**

 * @ProjectName: cutter-point

 * @Package: queue.fqueue.vo

 * @ClassName: EventVo

 * @Author: xiaof

 * @Description: ${description}

 * @Date: 2019/6/11 10:30

 * @Version: 1.0

 */

public interface EventVo {

    public void doOperater();

}

package queue.fqueue.vo;

import java.io.Serializable;

/**

 * @ProjectName: cutter-point

 * @Package: queue.fqueue.vo

 * @ClassName: TempVo

 * @Author: xiaof

 * @Description: ${description}

 * @Date: 2019/6/11 10:18

 * @Version: 1.0

 */

public class TempVo implements Serializable, EventVo {

    private String name;

    public String getName() {

        return name;

    }

    public void setName(String name) {

        this.name = name;

    }

    @Override

    public String toString() {

        return "TempVo{name='" + name + "'}";

    }

    @Override

    public void doOperater() {

        System.out.println(name + " : say hello fqueue!");

    }

}

package queue.fqueue;

import net.apexes.fqueue.FQueue;

import queue.fqueue.vo.TempVo;

import java.io.*;

/**

 * @ProjectName: cutter-point

 * @Package: queue.fqueue

 * @ClassName: FqueueProducter

 * @Author: xiaof

 * @Description: ${description}

 * @Date: 2019/6/11 10:36

 * @Version: 1.0

 */

public class FqueueProducter implements Runnable {

    private FQueue fQueue;

    public FqueueProducter(FQueue fQueue) {

        this.fQueue = fQueue;

    }

    @Override

    public void run() {

        while(true) {

            try {

                Thread.sleep(2000);

                TempVo tempVo = new TempVo();

                tempVo.setName(Thread.currentThread().getName() + ",time is:" + System.currentTimeMillis());

                //序列化为字节

                OutputStream arrayOutputStream = new ByteArrayOutputStream();

                ObjectOutputStream objectOutputStream = new ObjectOutputStream(arrayOutputStream);

                objectOutputStream.writeObject(tempVo);

                arrayOutputStream.flush();

                fQueue.add(((ByteArrayOutputStream) arrayOutputStream).toByteArray());

            } catch (InterruptedException e) {

                e.printStackTrace();

            } catch (IOException e) {

                e.printStackTrace();

            }

        }

    }

}

package queue.fqueue;

import net.apexes.fqueue.FQueue;

import queue.fqueue.vo.EventVo;

import java.io.ByteArrayInputStream;

import java.io.IOException;

import java.io.ObjectInputStream;

/**

 * @ProjectName: cutter-point

 * @Package: queue.fqueue

 * @ClassName: FqueueProducter

 * @Author: xiaof

 * @Description: ${description}

 * @Date: 2019/6/11 9:40

 * @Version: 1.0

 */

public class FqueueConsume implements Runnable {

    private FQueue fQueue;

    public FqueueConsume(FQueue fQueue) {

        this.fQueue = fQueue;

    }

    @Override

    public void run() {

        while(true) {

            byte bytes[] = fQueue.poll();

            //反序列化对象

            if(bytes == null || bytes.length <= 0) {

                Thread.yield();

                continue;

            }

            ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);

            try {

                ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);

                EventVo eventVo = (EventVo) objectInputStream.readObject();

                eventVo.doOperater();

            } catch (IOException e) {

                e.printStackTrace();

            } catch (ClassNotFoundException e) {

                e.printStackTrace();

            }

        }

    }

}

测试代码：

@Test

    public void test3() throws IOException, FileFormatException, InterruptedException {

        FQueue queue1 = new FQueue("db1");

        //读写取数据

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

            System.out.println("输出测试" + i);

            FqueueProducter producter = new FqueueProducter(queue1);

            Thread t = new Thread(producter);

            t.start();

        }

        //读写取数据

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

            System.out.println("输出测试" + i);

            FqueueConsume fqueueConsume = new FqueueConsume(queue1);

            Thread t = new Thread(fqueueConsume);

            t.setDaemon(true);

            t.start();

        }

        while(true) {

            Thread.sleep(1000);

        }

    }

效果展示：

巴特西

【并发】6、借助FQueue 实现多线程生产消费队列

最新文章

热门文章