目录
🍑前言:
🍑使用ConcurrentHashMap在高并发场景下进行数据操作的示例
🍑我们使用了100个线程来添加元素，这可能会导致线程数过多而影响性能,因此，可以通过使用线程池来管理线程数量
🍑 我们使用了并行流来并发处理任务，但是这种方式会占用大量的CPU资源，可能会导致其他线程无法正常运行。因此，我们可以使用分治策略来将数据划分成多个小批量进行处理，以避免CPU资源的浪费
🍑我们使用了线程池和分治策略来优化程序效率。然而，当元素数量较大时，遍历数据仍然会消耗大量的时间和资源。为了进一步提高程序效率，我们可以使用并发流和批量添加元素的方式来处理数据。
🍑我们使用了分治策略、线程池和并发流来优化程序效率。尽管这些方法可以提高程序效率，但当元素数量非常大时，仍然可能会遇到性能瓶颈。为了进一步提升程序性能，使用Spark框架进行并行计算
🍑SparkSession sparkSession = SparkSession.builder().master("local[*]").appName("Demo").getOrCreate();是干什么的?
🍑Apache Cassandra分布式数据库进行分布式存储和查询
🍑Redis作为分布式缓存
🍑前言:
主要是由简到深来讲解关于ConcurrentHashMap这个知识点的应用,希望对你有帮助,沉淀自己应用的笔记
🍑使用ConcurrentHashMap在高并发场景下进行数据操作的示例
import java.util.concurrent.ConcurrentHashMap;public class Demo {public static void main(String[] args) {ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();// 创建100个线程，每个线程向map中添加10000个元素for (int i = 0; i < 100; i++) {new Thread(() -> {for (int j = 0; j < 10000; j++) {String key = "key" + j;Integer value = j;map.put(key, value);}}).start();}// 等待所有线程执行完成try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}// 输出map的大小System.out.println("map size: " + map.size());}
}
- 这个示例中，我们创建了100个线程，并使用ConcurrentHashMap向其中添加10000个元素。由于ConcurrentHashMap是线程安全的，所以可以在高并发场景下保证数据的一致性和完整性。最终，我们输出map的大小，可以看到其大小为1000000，符合预期。
🍑我们使用了100个线程来添加元素，这可能会导致线程数过多而影响性能,因此，可以通过使用线程池来管理线程数量
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;public class Demo {public static void main(String[] args) {ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();// 创建10个线程的线程池ExecutorService executorService = Executors.newFixedThreadPool(10);// 使用并行流并发处理任务IntStream.range(0, 100000).parallel().forEach(i -> {String key = "key" + i;Integer value = i;map.put(key, value);});// 关闭线程池executorService.shutdown();// 输出map的大小System.out.println("map size: " + map.size());}
}
- 我们创建了一个包含10个线程的线程池，并使用并行流处理任务。在实际应用中，应该根据实际情况选择线程池的大小和并行度。最终，我们输出map的大小，可以看到其大小为100000，符合预期。
🍑 我们使用了并行流来并发处理任务，但是这种方式会占用大量的CPU资源，可能会导致其他线程无法正常运行。因此，我们可以使用分治策略来将数据划分成多个小批量进行处理，以避免CPU资源的浪费
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;public class Demo {public static void main(String[] args) {ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();// 创建10个线程的线程池ExecutorService executorService = Executors.newFixedThreadPool(10);// 将100000个元素划分成100个小批量，每个小批量包含1000个元素int batchSize = 1000;IntStream.range(0, 100).forEach(i -> {int start = i * batchSize;int end = Math.min((i + 1) * batchSize, 100000);executorService.submit(() -> {IntStream.range(start, end).forEach(j -> {String key = "key" + j;Integer value = j;map.put(key, value);});});});// 关闭线程池executorService.shutdown();// 输出map的大小System.out.println("map size: " + map.size());}
}
- 在这个示例中，我们将100000个元素划分成100个小批量，每个小批量包含1000个元素。对于每个小批量，我们将其交给线程池中的线程来处理。由于使用了分治策略，避免了CPU资源的浪费，可以提高程序的效率。最终，我们输出map的大小，可以看到其大小为100000，符合预期。
🍑我们使用了线程池和分治策略来优化程序效率。然而，当元素数量较大时，遍历数据仍然会消耗大量的时间和资源。为了进一步提高程序效率，我们可以使用并发流和批量添加元素的方式来处理数据。
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;public class Demo {public static void main(String[] args) {ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();// 创建10个线程的线程池ExecutorService executorService = Executors.newFixedThreadPool(10);// 将100000个元素划分成100个小批量，每个小批量包含1000个元素int batchSize = 1000;IntStream.range(0, 100).forEach(i -> {int start = i * batchSize;int end = Math.min((i + 1) * batchSize, 100000);executorService.submit(() -> {ConcurrentHashMap<String, Integer> batchMap = new ConcurrentHashMap<>();IntStream.range(start, end).forEach(j -> {String key = "key" + j;Integer value = j;batchMap.put(key, value);});map.putAll(batchMap);});});// 关闭线程池executorService.shutdown();// 输出map的大小System.out.println("map size: " + map.size());}
}
- 在这个示例中，我们使用了并发流来遍历1000个元素，并使用ConcurrentHashMap批量添加元素。每个小批量的元素都是由一个线程处理，然后将结果合并到总的map中。这种方式可以有效地减少数据遍历的时间和资源消耗。最终，我们输出map的大小，可以看到其大小为100000，符合预期。
🍑我们使用了分治策略、线程池和并发流来优化程序效率。尽管这些方法可以提高程序效率，但当元素数量非常大时，仍然可能会遇到性能瓶颈。为了进一步提升程序性能，使用Spark框架进行并行计算
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.sql.SparkSession;import java.util.concurrent.ConcurrentHashMap;public class Demo {public static void main(String[] args) {// 创建SparkSession和JavaSparkContextSparkSession sparkSession = SparkSession.builder().master("local[*]").appName("Demo").getOrCreate();JavaSparkContext sparkContext = new JavaSparkContext(sparkSession.sparkContext());// 创建100万个元素的JavaRDDint size = 1000000;JavaRDD<Integer> rdd = sparkContext.parallelize(java.util.Arrays.asList(new Integer[size]), 10).mapPartitions(iterator -> {ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();while (iterator.hasNext()) {Integer value = iterator.next();String key = "key" + value;map.put(key, value);}return java.util.Arrays.asList(map).iterator();}).reduce((Function2<ConcurrentHashMap<String, Integer>, ConcurrentHashMap<String, Integer>, ConcurrentHashMap<String, Integer>>) (m1, m2) -> {m1.putAll(m2);return m1;});// 输出RDD中的元素数量System.out.println("RDD size: " + rdd.count());}
}
🍑SparkSession sparkSession = SparkSession.builder().master("local[*]").appName("Demo").getOrCreate();是干什么的?
- 这段代码是在创建一个SparkSession对象，它是Spark 2.x版本中的入口点，可以让我们与Spark集群进行交互。在这里，我们使用了
builder()方法来构建一个SparkSession.Builder对象，并通过master()方法指定了本地运行模式(local[*])和应用程序名称(Demo)。- 其中，
local[*]表示使用所有可用的核心数，在本地模式下运行Spark应用程序，这样可以方便我们进行测试和开发。在真实生产环境中，需要指定Spark集群的URL或者其他连接信息。- 最后，使用
getOrCreate()方法来获取或创建一个SparkSession对象。如果已经存在一个- 可用的
SparkSession对象，则直接返回该对象；否则，会新建一个SparkSession对象。
🍑Apache Cassandra分布式数据库进行分布式存储和查询
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;public class Demo {public static void main(String[] args) {// 创建一个Cassandra集群对象，和一个Session对象Cluster cluster = Cluster.builder().addContactPoint("127.0.0.1").build();Session session = cluster.connect();// 创建一个名为my_keyspace的keyspaceString keyspace_name = "my_keyspace";session.execute(String.format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class':'SimpleStrategy', 'replication_factor':1};", keyspace_name));// 创建一个名为my_table的tableString table_name = "my_table";session.execute(String.format("CREATE TABLE IF NOT EXISTS %s.%s (key text PRIMARY KEY, value int);", keyspace_name, table_name));// 向表中添加100万个元素ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();for(int i = 0; i < 1000000; i++) {String key = "key" + UUID.randomUUID().toString();Integer value = i;map.put(key, value);}for(String key: map.keySet()) {Integer value = map.get(key);session.execute(String.format("INSERT INTO %s.%s (key,value) VALUES ('%s',%d);", keyspace_name, table_name, key, value));}// 查询表中的所有元素ResultSet resultSet = session.execute(String.format("SELECT * FROM %s.%s;", keyspace_name, table_name));ConcurrentHashMap<String, Integer> resultMap = new ConcurrentHashMap<>();for(Row row: resultSet) {String key = row.getString(0);Integer value = row.getInt(1);resultMap.put(key, value);}// 输出查询结果System.out.println("result size: " + resultMap.size());// 关闭Session和Cluster对象session.close();cluster.close();}
}
- 示例中，我们首先创建了一个Cassandra集群对象和一个Session对象，然后声明一个名为
my_keyspace的keyspace和一个名为my_table的table。接着，我们向表中添加100万个元素，并在查询结果中验证数据的完整性。 - 使用分布式数据库可以轻松地将数据存储到多个节点上，并且可以支持复杂的查询和聚合操作。同时，使用分布式数据库还可以提高系统的可伸缩性和容错性，适用于处理大规模数据的场景。
🍑Redis作为分布式缓存
我们还可以使用分布式缓存来提高程序的性能。具体来说，可以使用一个高性能的内存访问数据库，如Redis或Memcached，将结果缓存起来，以减少每次查询时要读取磁盘或远程服务器的数据量。
import redis.clients.jedis.Jedis;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;public class Demo {public static void main(String[] args) {// 连接本地Redis服务Jedis jedis = new Jedis("localhost");// 创建一个名为my_table的HashMapMap<String, Integer> map = new HashMap<>();for(int i = 0; i < 1000000; i++) {String key = "key" + UUID.randomUUID().toString();Integer value = i;map.put(key, value);}// 将HashMap中的元素存入Redis缓存中for(String key: map.keySet()) {Integer value = map.get(key);jedis.set(key, String.valueOf(value));}// 查询缓存中的所有元素Map<String, String> resultMap = jedis.hgetAll("*");// 输出查询结果System.out.println("result size: " + resultMap.size());// 关闭Jedis连接jedis.close();}
}
- 示例中，我们首先连接到一个名为
localhost的Redis服务，并声明一个名为my_table的HashMap。然后，我们向HashMap中添加100万个元素，并使用Redis将元素存储到缓存中。接着，我们查询缓存中的所有元素，并验证数据的完整性。 - 使用分布式缓存可以将数据存储在内存中，减少了磁盘或远程服务器的访问次数，从而提高了程序的效率和响应速度。如果数据量较小，可以直接将数据存储在单机内存中，使用本地缓存来实现类似的功能。
