Java多线程分批发送消息的小例子

需求:

假设有10万个用户,现在节假日做活动,需要给每个用户发送一条活动短信,为了提高程序的效率,建议使用多线程分批发送.

这里值得注意的是:

每开一个线程都会占用CPU的资源,所以线程根据所需要的条数来决定就好,避免浪费,我们用的是一个小例子,只是说明了多线程处理提高了效率,实际的大规模场景中不建议使用,可选消息中间件来轮询处理.

Demo示例:

这里为简化展示,模拟示例10条内容,用分页处理的方式来分批发送消息,每批为2条:

结构如下:

entity

———– UserEntity

batch

———– BatchSms

utils
1
|----------- ListUtils

相关类如下:

UserEntity:

package entity;

public class UserEntity {
    private String userId;
    private String userName;


    public UserEntity(String userId, String userName) {
        this.userId = userId;
        this.userName = userName;
    }

    public String getUserId() {
        return userId;
    }

    public void setUserId(String userId) {
        this.userId = userId;
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    @Override
    public String toString() {
        return "UserEntity{" +
                "userId='" + userId + '

BatchSms:

1
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58 package batch;

import entity.UserEntity;
import utils.ListUtils;

import java.util.ArrayList;
import java.util.List;

class UserSendThread implements Runnable {
    private List<UserEntity> listUser;

    public UserSendThread(List<UserEntity> listUser) {
        this.listUser = listUser;
    }

    public void run() {
        for (UserEntity userEntity : listUser) {
            System.out.println(Thread.currentThread().getName() + " " + userEntity.toString());
        }

    }
}

public class BatchSms {

    public static void main(String[] args) {

        // 1. 初始化数据
        List<UserEntity> list = initUser();

        // 2.定义每个线程分批发送大小
        int userCount = 2;

        // 3.计算每个线程需要分批跑的数据
        List<List<UserEntity>> splitList = ListUtils.splitList(initUser(), userCount);

        for (int i = 0; i < splitList.size(); i++) {
            List<UserEntity> list1 = splitList.get(i);
            UserSendThread userSendThread = new UserSendThread(list1);
            // 4.分批发送
            Thread thread = new Thread(userSendThread, "线程" + i);
            thread.start();
            System.out.println();
        }
    }

    private static List<UserEntity> initUser() {

        List<UserEntity> list = new ArrayList<UserEntity>();

        for (int i = 0; i < 10; i++) {
            list.add(new UserEntity("userid:" + i, "username" + i));
        }

        return list;
    }
}



ListUtils:

1
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 package utils;

import java.util.ArrayList;
import java.util.List;

public class ListUtils {
    /**
     *
     * @methodDesc: 功能描述:(list 集合分批切割)
     * @param: @param
     *             list
     * @param: @param
     *             pageSize
     * @param: @return
     * @returnType:@param list 切割集合
     * @returnType:@param pageSize 分页长度
     * @returnType:@return List<List<T>> 返回分页数据
     */

    static public<T> List<List<T>> splitList(List<T> list, int pageSize) {
        int listSize = list.size();
        int page = (listSize + (pageSize - 1)) / pageSize;
        List<List<T>>listArray = new ArrayList<List<T>>();
        for (int i = 0; i<page; i++) {
            List<T>subList = new ArrayList<T>();
            for (int j = 0; j<listSize; j++) {
                int pageIndex = ((j + 1) + (pageSize - 1)) / pageSize;
                if (pageIndex == (i + 1)) {
                    subList.add(list.get(j));
                }
                if ((j + 1) == ((j + 1) * pageSize)) {
                    break;
                }
            }
            listArray.add(subList);
        }
        return listArray;
    }
}


启动后运行结果如下:


' +
                ", userName='" + userName + '

BatchSms:

1
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58 package batch;

import entity.UserEntity;
import utils.ListUtils;

import java.util.ArrayList;
import java.util.List;

class UserSendThread implements Runnable {
    private List<UserEntity> listUser;

    public UserSendThread(List<UserEntity> listUser) {
        this.listUser = listUser;
    }

    public void run() {
        for (UserEntity userEntity : listUser) {
            System.out.println(Thread.currentThread().getName() + " " + userEntity.toString());
        }

    }
}

public class BatchSms {

    public static void main(String[] args) {

        // 1. 初始化数据
        List<UserEntity> list = initUser();

        // 2.定义每个线程分批发送大小
        int userCount = 2;

        // 3.计算每个线程需要分批跑的数据
        List<List<UserEntity>> splitList = ListUtils.splitList(initUser(), userCount);

        for (int i = 0; i < splitList.size(); i++) {
            List<UserEntity> list1 = splitList.get(i);
            UserSendThread userSendThread = new UserSendThread(list1);
            // 4.分批发送
            Thread thread = new Thread(userSendThread, "线程" + i);
            thread.start();
            System.out.println();
        }
    }

    private static List<UserEntity> initUser() {

        List<UserEntity> list = new ArrayList<UserEntity>();

        for (int i = 0; i < 10; i++) {
            list.add(new UserEntity("userid:" + i, "username" + i));
        }

        return list;
    }
}



ListUtils:

1
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 package utils;

import java.util.ArrayList;
import java.util.List;

public class ListUtils {
    /**
     *
     * @methodDesc: 功能描述:(list 集合分批切割)
     * @param: @param
     *             list
     * @param: @param
     *             pageSize
     * @param: @return
     * @returnType:@param list 切割集合
     * @returnType:@param pageSize 分页长度
     * @returnType:@return List<List<T>> 返回分页数据
     */

    static public<T> List<List<T>> splitList(List<T> list, int pageSize) {
        int listSize = list.size();
        int page = (listSize + (pageSize - 1)) / pageSize;
        List<List<T>>listArray = new ArrayList<List<T>>();
        for (int i = 0; i<page; i++) {
            List<T>subList = new ArrayList<T>();
            for (int j = 0; j<listSize; j++) {
                int pageIndex = ((j + 1) + (pageSize - 1)) / pageSize;
                if (pageIndex == (i + 1)) {
                    subList.add(list.get(j));
                }
                if ((j + 1) == ((j + 1) * pageSize)) {
                    break;
                }
            }
            listArray.add(subList);
        }
        return listArray;
    }
}


启动后运行结果如下:


' +
                '}';
    }
}

BatchSms:

package batch;

import entity.UserEntity;
import utils.ListUtils;

import java.util.ArrayList;
import java.util.List;

class UserSendThread implements Runnable {
    private List<UserEntity> listUser;

    public UserSendThread(List<UserEntity> listUser) {
        this.listUser = listUser;
    }

    public void run() {
        for (UserEntity userEntity : listUser) {
            System.out.println(Thread.currentThread().getName() + " " + userEntity.toString());
        }

    }
}

public class BatchSms {

    public static void main(String[] args) {

        // 1. 初始化数据
        List<UserEntity> list = initUser();

        // 2.定义每个线程分批发送大小
        int userCount = 2;

        // 3.计算每个线程需要分批跑的数据
        List<List<UserEntity>> splitList = ListUtils.splitList(initUser(), userCount);

        for (int i = 0; i < splitList.size(); i++) {
            List<UserEntity> list1 = splitList.get(i);
            UserSendThread userSendThread = new UserSendThread(list1);
            // 4.分批发送
            Thread thread = new Thread(userSendThread, "线程" + i);
            thread.start();
            System.out.println();
        }
    }

    private static List<UserEntity> initUser() {

        List<UserEntity> list = new ArrayList<UserEntity>();

        for (int i = 0; i < 10; i++) {
            list.add(new UserEntity("userid:" + i, "username" + i));
        }

        return list;
    }
}

ListUtils:

package utils;

import java.util.ArrayList;
import java.util.List;

public class ListUtils {
    /**
     *
     * @methodDesc: 功能描述:(list 集合分批切割)
     * @param: @param
     *             list
     * @param: @param
     *             pageSize
     * @param: @return
     * @returnType:@param list 切割集合
     * @returnType:@param pageSize 分页长度
     * @returnType:@return List<List<T>> 返回分页数据
     */

    static public<T> List<List<T>> splitList(List<T> list, int pageSize) {
        int listSize = list.size();
        int page = (listSize + (pageSize - 1)) / pageSize;
        List<List<T>>listArray = new ArrayList<List<T>>();
        for (int i = 0; i<page; i++) {
            List<T>subList = new ArrayList<T>();
            for (int j = 0; j<listSize; j++) {
                int pageIndex = ((j + 1) + (pageSize - 1)) / pageSize;
                if (pageIndex == (i + 1)) {
                    subList.add(list.get(j));
                }
                if ((j + 1) == ((j + 1) * pageSize)) {
                    break;
                }
            }
            listArray.add(subList);
        }
        return listArray;
    }
}

启动后运行结果如下: